Flink 编程模型

Flink 环境准备

• Flink 编写程序需要依赖 Java——JDK
• 项目使用 Maven 管理依赖 ——Maven
• 开发工具使用 IDEA——IntelliJ IDEA

JDK 8

https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

Maven

http://maven.apache.org/download.cgi

IntelliJ IDEA

https://www.jetbrains.com/idea/download/#section=windows

下载 Flink

https://flink.apache.org/downloads.html

安装 Scala Plugins

点击 File -> Settings 菜单，或 Ctrl + Alt + S 快捷键。打开设置面板。并切换到 Plugins 插件视图
搜索 Scala 点击 Install

Flink 项目模版

基于 Java 的项目模版

Flink WordCount Java 源码 GitHub

在命令行使用 maven 创建 Flink 项目

mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.8.3

根据提示输入 groupId、artifactId

• groupId：com.qst（所在公司、学校、组织官网网址的反写）
• artifactId：wordcount-java（项目名称）

项目目录结构

• 使用 mvn 命令创建项目后我们会得到一个如下结构的项目目录

编译项目

• 在项目所在目录执行 mvn clean package 命令对项目进行编译
• 这时 maven 会下载 Flink 项目需要的依赖包并编译项目
• 编译完成后产生一个 target/<artifact-id>-<version>.jar 文件

基于 Scala 的项目模版

mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-scala \
-DarchetypeVersion=1.8.3

Flink WordCount

Flink WordCount scala 源码 GitHub

创建 WrodCount 项目

在命令行使用 maven 创建 Flink 项目

mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-scala \
-DarchetypeVersion=1.8.3

根据提示在输入 groupId、artifactId

• groupId：com.qst（所在公司、学校、组织官网网址的反写）
• artifactId：wordcount-scala（项目名称）
• 其它选项使用默认值

将项目导入 IDEA

• 在 IDEA 中将 flink-wordcount 项目导入
• 选择 Import Project 找到 wordcount-scala 所在目录将项目导入 IDEA

开发 WordCount 程序

第一步：设定执行环境

• 运行 Flink 程序的第一步就是获得相应的执行环境，执行环境决定了程序在什么环境执行（本地 / 集群）
• 不同的运行环境也决定了程序的类型

• 批处理 ExecutionEnvironment
• 流处理 StreamExecutionEnvironment

第二步：指定数据源 读取数据

• 定义执行环境后需要获得需要处理的数据，将外部数据转换成 DateStream 或 DataSet
  如下方法读取所示使用 readTextFile() 方法读取文件中的数据并转换成 DataStream 数据集

  val source = env.readTextFile("/word.txt")

第三步：对数据集执行转换操作

• Flink 中的 Transformation 操作通过不同的 Operator 来实现对数据的操作
• Operator 内部通过 Function 接口完成数据处理
• 在 DataStream API 和 DataSet API 中提供了大量的转换操作
  flatMap、map、filter、keyBy

source.flatMap(line => line.toLowerCase.split("\\W+"))  //将文本转换成数组
.filter(_.length > 0)  //过滤空字符串
.map(word => (word, 1)) //转换成 key-value 接口
.keyBy(0) //按照指定字段（key）对数据进行分区
.sum(1) //执行求和运算

第四步：输出结果

经过转换后形成了最终结果，通常需要将结果数据输出到外部系统中

source.flatMap(line => line.toLowerCase.split("\\W+"))  //将文本转换成数组
.filter(_.length > 0)  //过滤空字符串
.map(word => (word, 1)) //转换成 key-value 接口
.keyBy(0) //按照指定字段（key）对数据进行分区
.sum(1) //执行求和运算
.print() //输出到控制台
//.writeAsText("/word_out.txt") //写入外部文件

第五步：触发程序执行

所有的计算逻辑完成之后，需要调用 StreamExecutionEnvironment 的 execute 方法来触发应用程序的执行

env.execute("Streaming Scala WordCount")

运行 & 编译 WordCount 程序

编译 WordCount 应用程序

• 在程序根目录执行 mvn clean package 命令进行编译
• 这时 maven 会下载 Flink 项目需要的依赖包并编译项目
• 编译完成后产生一个 target/<artifact-id>-<version>.jar 文件