Apache Flink实战(二) - 第一个Flink应用程序
动手搭建Flink的开发环境，快速使用Java和Scala语言开发第一个基于Flink的批处理和流式处理的应用程序。
相关源码
下载安装
brew install apache-flink
1 需求描述
Flink开发批处理应用程序
需求:词频统计(word count)
一个文件，统计文件中每个单词出现的次数
分隔符是\t
统计结果我们直接打印在控制台(生产上肯定是Sink到目的地)
实现:
Flink + Java
Flink + Scala
2 Java开发之环境准备
2.1 Maven构建
2.1.1 Requirements
唯一的要求是Maven 3.0.4（或更高版本）和 Java 8.x 安装
2.1.2 Create Projec
使用以下命令之一创建项目：
maven archetype
$ mvn archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.8.0
这允许您为新创建的项目命名。 它将以交互方式询问您groupId，artifactId和包名称。
Run the quickstart script
$ curl https://flink.apache.org/q/quickstart.sh | bash -s 1.8.0
2.1.3 Inspect Project
您的工作目录中将有一个新目录。 如果您使用了curl方法，则该目录称为quickstart。 否则，它具有artifactId的名称：
使用IDEA打开该项目即可!
示例项目是一个Maven项目，它包含两个类：StreamingJob和BatchJob是DataStream和DataSet程序的基本框架程序。
主要方法是程序的入口点，既可用于IDE测试/执行，也可用于正确部署。
建议将此项目导入IDE以进行开发和测试。 IntelliJ IDEA支持开箱即用的Maven项目。
不建议Eclipse
请注意：对于Flink，Java的默认JVM堆可能太小。 你必须手动增加它。在IntelliJ IDEA中，推荐的更改JVM选项的方法来自Help | 编辑自定义VM选项菜单 -Xmx800m
2.1.4 Build Project
如果要构建/打包项目，请转到项目目录并运行
mvn clean package
或者使用插件
您将找到包含应用程序的JAR文件，以及可能已作为依赖项添加到应用程序的连接器和库：
target / <artifact-id> - <version> .jar
注意：如果您使用与StreamingJob不同的类作为应用程序的主类/入口点，我们建议您相应地更改pom.xml文件中的mainClass设置。 这样，Flink可以从JAR文件运行应用程序，而无需另外指定主类。
3 公式型编程
- step 1 : set up the batch execution environment
- step 2 : Start with getting some data from the environment, like
env.readTextFile(textPath);
- step 3 : 开发业务逻辑的核心代码
transform the resulting DataSet using operations,like
.filter()
.flatMap()
.join()
.coGroup()
- step 4 : execute program
4 Flink批处理应用Java开发之功能实现
-
在相应的目录下建立文本
-
测试代码
-
成功读取
4.1 功能拆解
1）读取数据
Hello JavaEdge
2）每一行的数据按照指定的分隔符拆分
Hello
JavaEdge
3）为每一个单词赋上次数为1
(Hello,1)
(JavaEdge,1)
4) 合并操作
groupBy
- 代码
- 结果
5 Scala开发之环境准备
-
要求与Java环境基本类似,不再赘述
-
Maven构建
6 批处理应用Scala功能实现
- 实现代码
- 内容正确打印
7 实时处理应用Java开发之功能实现
- 代码
可能遇到拒绝连接,记得
nc -lk 9999
-
清除运行结果
-
socket数据
-
结果
9 Flink实时处理应用Java开发之代码重构
如何才能突破端口被限制呢，需要如下重构：
- 传入参数
args
- 获得参数
10 实时处理应用Scala开发之代码重构
11 开发过程中依赖的注意事项
Configuring Dependencies, Connectors, Libraries
每个Flink应用程序都依赖于一组Flink库。 至少，应用程序依赖于Flink API。 许多应用程序还依赖于某些连接器库（如Kafka，Cassandra等）。 运行Flink应用程序时（在分布式部署中或在IDE中进行测试），Flink运行时库也必须可用。
