Flink 程序结构 下篇

这次接着上次的话题继续分享：分区 key 的指定、输出结果和程序触发

（4） 分区 key 的指定

Flink 的某些转换算子，如 join、coGroup、groupBy 算子，需要先将 DataStream 或 DataSet 数据集转换成对应的 KeyedStream 或 GroupedDataSet，主要目的是将相同的 key 值的数据路由到相同的 pipeline 中，然后进行下一步的计算操作。

需要注意的是，Flink 并不是真正意义上的 转换成 key - value 操作，而是一种虚拟 key。

有两种指定方式

a. 根据字段位置指定

上一段示例代码

流式计算的 keyBy

env.fromElements(("a",1),("a",3),("b",2),("c",3))
 // 根据第一个字段重新分区，然后对第二个字段进行求和计算
 .keyBy(0)
 .sum(1)
 .print()

批量计算的 groupBy

env.fromElements(("a",1),("a",3),("b",2),("c",3))
   // 根据第一个字段重新分区，找到第二个字段下的最大值
   .groupBy(0)
   .max(1)
   .print()

b. 根据字段名称指定

要想根据名称指定，则 DataStream 中的数据结构类型必须是 Tuple 类 或者 POJOs 类。

使用 POJOs 类，可以使用字段名来指定

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 来指定

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 基本数据类型