这次接着上次的话题继续分享：分区 key 的指定、输出结果和程序触发
（4） 分区 key 的指定
Flink 的某些转换算子，如 join、coGroup、groupBy 算子，需要先将 DataStream 或 DataSet 数据集转换成对应的 KeyedStream 或 GroupedDataSet，主要目的是将相同的 key 值的数据路由到相同的 pipeline 中，然后进行下一步的计算操作。
需要注意的是，Flink 并不是真正意义上的 转换成 key - value 操作，而是一种虚拟 key。
有两种指定方式
a. 根据字段位置指定
上一段示例代码
流式计算的 keyBy
代码语言：javascript复制env.fromElements(("a",1),("a",3),("b",2),("c",3))
// 根据第一个字段重新分区，然后对第二个字段进行求和计算
.keyBy(0)
.sum(1)
.print()批量计算的 groupBy
代码语言：javascript复制env.fromElements(("a",1),("a",3),("b",2),("c",3))
// 根据第一个字段重新分区，找到第二个字段下的最大值
.groupBy(0)
.max(1)
.print()b. 根据字段名称指定
要想根据名称指定，则 DataStream 中的数据结构类型必须是 Tuple 类 或者 POJOs 类。
使用 POJOs 类，可以使用字段名来指定
代码语言：javascript复制case class Person(name:String,age:Int)
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.fromElements(Person("zhangsan",23),Person("lisi",27),Person("wangwu",29))
.keyBy("name")
.max(1)
.print()
env.execute("job")使用 Tuple 结构，可以使用 _1 来指定
代码语言：javascript复制case class Person(name:String,age:Int)
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.fromElements(("zhangsan",1),("lisi",3),("wangwu",8))
.keyBy("_1")
.max(1)
.print()
env.execute("job")（5）输出结果
数据集经过转换之后，形成最终的结果数据集，一般结果会写入到外部系统或者打印到控制台。
例如基于文件输出 writeAsText()，基于控制台输出 print() 等。
同时 Flink 在系统中定义了大量的 Connector，方便用户和外部系统交互，用户可以直接调用 addSink() 添加输出系统定义的 DataSink 类算子。
（6）程序触发
所有计算逻辑定义好之后，需要调用 ExecutionEnvironment 的 execute 方法来触发应用程序的执行。
流式的应用需要显示的调用 execute() 来触发执行，批量计算则不用显示调用，输出算子已经包含对execute的调用了。
到了这儿，Flink 程序结构部分基本讲完了，来温习一下一个完整的Flink程序是哪些部分组成的：
1、执行环境，ExecutionEnvironment
2、初始化数据
3、数据转换操作
4、（可选）分区 key 指定
5、输出结果
6、触发执行（流式计算需要，DataSet Api 不需要）
下一次，我们会讲 Flink 基本数据类型
