Flink 原理、实战与性能优化 (22):Flink 编程模型 3.3

阅读数:7 2019 年 12 月 11 日 20:42

Flink原理、实战与性能优化(22):Flink编程模型 3.3

(Flink 程序结构)

内容简介
这是一部以实战为导向,能指导读者零基础掌握 Flink 并快速完成进阶的著作,从功能、原理、实战和调优等 4 个维度循序渐进地讲解了如何利用 Flink 进行分布式流式应用开发。作者是该领域的资深专家,现就职于第四范式,曾就职于明略数据。
全书一共 10 章,逻辑上可以分为三个部分:
第一部分(第 1~2 章)
主要介绍了 Flink 的核心概念、特性、应用场景、基本架构,开发环境的搭建和配置,以及源代码的编译。
第二部分(第 3~9 章)
详细讲解了 Flink 的编程范式,各种编程接口的功能、应用场景和使用方法,以及核心模块和组件的原理和使用。
第三部分(第 10 章)
重点讲解了 Flink 的监控和优化,参数调优,以及对反压、Checkpoint 和内存的优化。

和其他分布式处理引擎一样,Flink 应用程序也遵循着一定的编程模式。不管是使用 DataStream API 还是 DataSet API 基本具有相同的程序结构,如代码清单 3-1 所示。通过流式计算的方式实现对文本文件中的单词数量进行统计,然后将结果输出在给定路径中。

代码清单 3-1 Streaming WordCount 实例代码
复制代码
package com.realtime.flink.streaming
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecution
Environment, _}
object WordCount {
def main(args: Array[String]) {
// 第一步:设定执行环境设定
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 第二步:指定数据源地址,读取输入数据
val text = env.readTextFile("file:///path/file")
// 第三步:对数据集指定转换操作逻辑
val counts: DataStream[(String, Int)] = text
.flatMap(_.toLowerCase.split(" "))
.filter(_.nonEmpty)
.map((_, 1))
.keyBy(0)
.sum(1)
// 第四步:指定计算结果输出位置
if (params.has("output")) {
counts.writeAsText(params.get("output"))
} else {
println("Printing result to stdout. Use --output to specify output path.")
counts.print()
}
// 第五步:指定名称并触发流式任务
env.execute("Streaming WordCount")
}
}

整个 Flink 程序一共分为 5 步,分别为设定 Flink 执行环境、创建和加载数据集、对数据集指定转换操作逻辑、指定计算结果输出位置、调用 execute 方法触发程序执行。对于所有的 Flink 应用程序基本都含有这 5 个步骤,下面将详细介绍每个步骤。

1. Execution Environment

运行 Flink 程序的第一步就是获取相应的执行环境,执行环境决定了程序执行在什么环境(例如本地运行环境或者集群运行环境)中。同时不同的运行环境决定了应用的类型,批量处理作业和流式处理作业分别使用的是不同的 Execution Environment。例如 StreamExecutionEnvironment 是用来做流式数据处理环境,ExecutionEnvironment 是批

量数据处理环境。可以使用三种方式获取 Execution Environment,例如 StreamExecutionEnvirenment。

复制代码
// 设定 Flink 运行环境,如果在本地启动则创建本地环境,如果是在集群上启动,则创建集群环境
StreamExecutionEnvironment.getExecutionEnvironment
// 指定并行度创建本地执行环境
StreamExecutionEnvironment.createLocalEnvironment(5)
// 指定远程 JobManagerIP 和 RPC 端口以及运行程序所在 jar 包及其依赖包
StreamExecutionEnvironment.createRemoteEnvironment("JobManagerHost",6021,5,"/user/application.jar")

其中第三种方式可以直接从本地代码中创建与远程集群的 Flink JobManager 的 RPC 连接,通过指定应用程序所在的 Jar 包,将运行程序远程拷贝到 JobManager 节点上,然后将 Flink 应用程序运行在远程的环境中,本地程序相当于一个客户端。

和 StreamExecutionEnvironment 构建过程一样,开发批量应用需要获取 ExecutionEnvironment 来构建批量应用开发环境,如以下代码实例通过调用 ExecutionEnvironment 的静态方法来获取批计算环境。

复制代码
// 设定 Flink 运行环境,如果在本地启动则创建本地环境,如果是在集群上启动,则创建集群环境
ExecutionEnvironment.getExecutionEnvironment
// 指定并行度创建本地执行环境
ExecutionEnvironment.createLocalEnvironment(5)
// 指定远程 JobManagerIP 和 RPC 端口以及运行程序所在 jar 包及其依赖包
ExecutionEnvironment.createRemoteEnvironment("JobManagerHost",6021,5,"/user/application.jar")

针对 Scala 和 Java 不同的编程语言环境,Flink 分别制定了不同的语言同时分别定义了不同的 Execution Environment 接口。StreamExecutionEnvironment Scala 开发接口在 org.apache.flink.streaming.api.scala 包中,Java 开发接口在 org.apache.flink.streaming.api.java 包中;ExecutionEnvironment Scala 接口在 org.apache.flink.api.scala 包中,Java 开发接口则在 org.apache.flink.api.java 包中。用户使用不同语言开发 Flink 应用时需要引入不同环境对应的执行环境。

2. 初始化数据

创建完成 ExecutionEnvironment 后,需要将数据引入到 Flink 系统中。Execution-Environment 提供不同的数据接入接口完成数据的初始化,将外部数据转换成 DataStream或 DataSet数据集。如以下代码所示,通过调用 readTextFile() 方法读取 file:///pathfile 路径中的数据并转换成 DataStream数据集。

复制代码
val text:DataStream[String] = env.readTextFile("file:///path/file")

通过读取文件并转换为 DataStream[String] 数据集,这样就完成了从本地文件到分布式数据集的转换,同时在 Flink 中提供了多种从外部读取数据的连接器,包括批量和实时的数据连接器,能够将 Flink 系统和其他第三方系统连接,直接获取外部数据。

3. 执行转换操作

数据从外部系统读取并转换成 DataStream 或者 DataSet 数据集后,下一步就将对数据集进行各种转换操作。Flink 中的 Transformation 操作都是通过不同的 Operator 来实现,每个 Operator 内部通过实现 Function 接口完成数据处理逻辑的定义。在 DataStream API 和 DataSet API 提供了大量的转换算子,例如 map、flatMap、filter、keyBy 等,用户只需要定义每种算子执行的函数逻辑,然后应用在数据转换操作 Dperator 接口中即可。如下代码实现了对输入的文本数据集通过 FlatMap 算子转换成数组,然后过滤非空字段,将每个单词进行统计,得到最后的词频统计结果。

复制代码
val counts: DataStream[(String, Int)] = text
.flatMap(_.toLowerCase.split(" "))// 执行 FlatMap 转换操作
.filter(_.nonEmpty)// 执行 Filter 操作过滤空字段
.map((_, 1))// 执行 map 转换操作,转换成 key-value 接口
.keyBy(0)// 按照指定 key 对数据重分区
.sum(1)// 执行求和运算操作

在上述代码中,通过 Scala 接口处理数据,极大地简化数据处理逻辑的定义,只需要通过传入相应 Lambada 计算表达式,就能完成 Function 定义。特殊情况下用户也可以通过实现 Function 接口来完成定义数据处理逻辑。然后将定义好的 Function 应用在对应的算子中即可。Flink 中定义 Funciton 的计算逻辑可以通过如下几种方式完成定义。

(1)通过创建 Class 实现 Funciton 接口

Flink 中提供了大量的函数供用户使用,例如以下代码通过定义 MyMapFunction Class 实现 MapFunction 接口,然后调用 DataStream 的 map() 方法将 MyMapFunction 实现类传入,完成对实现将数据集中字符串记录转换成大写的数据处理。

复制代码
val dataStream: DataStream[String] = env.fromElements("hello", "flink")
dataStream.map(new MyMapFunction)
class MyMapFunction extends MapFunction[String, String] {
override def map(t: String): String = {
t.toUpperCase()
}
}

(2)通过创建匿名类实现 Funciton 接口

除了以上单独定义 Class 来实现 Function 接口之处,也可以直接在 map() 方法中创建匿名实现类的方式定义函数计算逻辑。

复制代码
val dataStream: DataStream[String] = env.fromElements("hello", "flink")
// 通过创建 MapFunction 匿名实现类来定义 Map 函数计算逻辑
dataStream.map(new MapFunction[String, String] {
// 实现对输入字符串大写转换
override def map(t: String): String = {
t.toUpperCase()
}
})

(3)通过实现 RichFunciton 接口

前面提到的转换操作都实现了 Function 接口,例如 MapFunction 和 FlatMap-Function 接口,在 Flink 中同时提供了 RichFunction 接口,主要用于比较高级的数据处理场景,RichFunction 接口中有 open、close、getRuntimeContext 和 setRuntimeContext 等方法来获取状态,缓存等系统内部数据。和 MapFunction 相似,RichFunction 子类中也有 RichMap-Function,如下代码通过实现 RichMapFunction 定义数据处理逻辑,具体的 RichFunction 的介绍读者可以参考后续章节中心介绍。

复制代码
// 定义匿名类实现 RichMapFunction 接口,完成对字符串到整形数字的转换
data.map (new RichMapFunction[String, Int] {
def map(in: String):Int = { in.toInt }
})

4. 分区 Key 指定

在 DataStream 数据经过不同的算子转换过程中,某些算子需要根据指定的 key 进行转换,常见的有 join、coGroup、groupBy 类算子,需要先将 DataStream 或 DataSet 数据集转换成对应的 KeyedStream 和 GroupedDataSet,主要目的是将相同 key 值的数据路由到相同的 Pipeline 中,然后进行下一步的计算操作。需要注意的是,在 Flink 中这种操作并不是真正意义上将数据集转换成 Key-Value 结构,而是一种虚拟的 key,目的仅仅是帮助后面的基于 Key 的算子使用,分区人 Key 可以通过两种方式指定:

(1)根据字段位置指定

在 DataStream API 中通过 keyBy() 方法将 DataStream 数据集根据指定的 key 转换成重新分区的 KeyedStream,如以下代码所示,对数据集按照相同 key 进行 sum() 聚合操作。

复制代码
val dataStream: DataStream[(String, Int)] = env.fromElements(("a", 1), ("c",
2))
// 根据第一个字段重新分区,然后对第二个字段进行求和运算
Val result = dataStream.keyBy(0).sum(1)

在 DataSet API 中,如果对数据根据某一条件聚合数据,对数据进行聚合时候,也需要对数据进行重新分区。如以下代码所示,使用 DataSet API 对数据集根据第一个字段作为 GroupBy 的 key,然后对第二个字段进行求和运算。

复制代码
val dataSet = env.fromElements(("hello", 1), ("flink", 3))
// 根据第一个字段进行数据重分区
val groupedDataSet:GroupedDataSet[(String,Int)] = dataSet.groupBy(0)
// 求取相同 key 值下第二个字段的最大值
groupedDataSet.max(1)

(2)根据字段名称指定

KeyBy 和 GroupBy 的 Key 除了能够通过字段位置来指定之外,也可以根据字段的名称来指定。使用字段名称需要 DataStream 中的数据结构类型必须是 Tuple 类或者 POJOs 类的。如以下代码所示,通过指定 name 字段名称来确定 groupby 的 key 字段。

复制代码
val personDataSet = env.fromElements(new Persion("Alex", 18),new
Persion("Peter", 43))
// 指定 name 字段名称来确定 groupby 字段
personDataSet.groupBy("name").max(1)

如果程序中使用 Tuple 数据类型,通常情况下字段名称从 1 开始计算,字段位置索引从 0 开始计算,以下代码中两种方式是等价的。

复制代码
val personDataStream = env.fromElements(("Alex", 18),("Peter", 43))
// 通过名称指定第一个字段名称
personDataStream.keyBy("_1")
// 通过位置指定第一个字段
personDataStream.keyBy(0)

如果在 Flink 中使用嵌套的复杂数据结构,可以通过字段名称指定 Key,例如:

复制代码
class CompelexClass(var nested: NestedClass, var tag: String) {
def this() { this(null, "") }
}
class NestedClass (
var id: Int,
tuple: (Long, Long, String)){
def this() { this(0, (0, 0, "")) }
}

通过调用“nested”获取整个 NestedClass 对象里所有的字段,调用“tag”获取 CompelexClass 中 tag 字段,调用“nested.id”获取 NestedClass 中的 id 字段,调用“nested.tuple._1”获取 NestedClass 中 tuple 元祖的第一个字段。由此可以看出,Flink 能够支持在复杂数据结构中灵活地获取字段信息,这也是非 Key-Value 的数据结构所具有的优势。

(3)通过 Key 选择器指定

另外一种方式是通过定义 Key Selector 来选择数据集中的 Key,如下代码所示,定义 KeySelector,然后复写 getKey 方法,从 Person 对象中获取 name 为指定的 Key。

复制代码
case class Person(name: String, age: Int)
val person= env.fromElements(Person("hello",1), Person("flink",4))
// 定义 KeySelector, 实现 getKey 方法从 case class 中获取 Key
override def getKey(person: Person): String = person.word
})

5. 输出结果

数据集经过转换操作之后,形成最终的结果数据集,一般需要将数据集输出在外部系统中或者输出在控制台之上。在 Flink DataStream 和 DataSet 接口中定义了基本的数据输出方法,例如基于文件输出 writeAsText(),基于控制台输出 print() 等。同时 Flink 在系统中定义了大量的 Connector,方便用户和外部系统交互,用户可以直接通过调用 addSink() 添加输出系统定义的 DataSink 类算子,这样就能将数据输出到外部系统。以下实例调用 DataStream API 中的 writeAsText() 和 print() 方法将数据集输出在文件和客户端中。

复制代码
// 将数据输出到文件中
counts.writeAsText("file://path/to/savefile")
// 将数据输出控制台
counts.print()

6. 程序触发

所有的计算逻辑全部操作定义好之后,需要调用 ExecutionEnvironment 的 execute() 方法来触发应用程序的执行,其中 execute() 方法返回的结果类型为 JobExecutionResult,里面包含了程序执行的时间和累加器等指标。需要注意的是,execute 方法调用会因为应用的类型有所不同,DataStream 流式应用需要显性地指定 execute() 方法运行程序,如果不调用则 Flink 流式程序不会执行,但对于 DataSet API 输出算子中已经包含对 execute() 方法的调用,则不需要显性调用 execute() 方法,否则会出现程序异常。

复制代码
// 调用 StreamExecutionEnvironment 的 execute 方法执行流式应用程序
env.execute("App Name");

Flink原理、实战与性能优化(22):Flink编程模型 3.3

购书地址 https://item.jd.com/12518733.html?dist=jd

评论

发布