深圳宝安高端网站建设报价,手机html编程软件app,西部数码网站管理助手搭建织梦,网站开发项目的简介前言
Flink 是大数据流计算引擎#xff0c;开发者通过程序语言开发一个 Flink 作业#xff0c;然后提交这个作业到服务端并执行#xff0c;以完成对大数据流的处理。
Flink 作业有一个基本骨架#xff0c;再复杂的 Flink 作业都离不开这个基本骨架#xff0c;了解作业的…前言
Flink 是大数据流计算引擎开发者通过程序语言开发一个 Flink 作业然后提交这个作业到服务端并执行以完成对大数据流的处理。
Flink 作业有一个基本骨架再复杂的 Flink 作业都离不开这个基本骨架了解作业的基本骨架有助于我们更快上手 Flink 作业的开发。
骨架结构
Flink 作业的基本骨架包含三个部分
创建Flink执行环境定义数据处理逻辑提交并执行Flink作业
创建Flink执行环境
Flink 执行环境被封装成 StreamExecutionEnvironment 对象通过该对象我们可以给Flink作业添加数据源、添加数据处理和输出逻辑、配置Flink作业的并行度、故障重启策略等参数。
定义数据处理逻辑
Flink是大数据流计算引擎本质上是对大数据的计算处理那么首先要解决的问题是数据从哪儿来
解决这个问题就是给Flink作业定义数据源数据源被抽象成了 SourceFunction 接口实现该接口重写 run 方法即可接收数据。
有了数据接下来就是声明要对这些数据做哪些处理对数据的处理被抽象成了 ProcessFunction 接口实现该接口重写 processElement 方法即可处理一条条数据。常见的数据处理操作有过滤、转换、聚合等。
数据计算以后还需要把计算结果给保存下来所以最后还需要一步数据汇 sink 操作把计算结果保存到数据存储引擎例如写入MySQL、Redis等存储引擎。
提交并执行Flink作业
到目前为止我们只描述了Flink的数据处理逻辑Flink作业不手动触发的前提下是不会自动执行的。
所以最后如果要让上述流程跑起来还需要手动提交并触发Flink作业。开发环境下可以直接在本地提交并执行生产环境一般是提交到Flink集群执行。
执行Flink作业对应的是 StreamExecutionEnvironment#execute 方法。
字数统计作业
这里以一个统计字数作业作为示例它被称作是 Flink 版的 Hello World虽然简单但是很好的体现了 Flink 作业的流程。
如下代码所示我们先是创建了Flink作业执行环境对象 StreamExecutionEnvironment然后定义了数据源监听本地的8888端口读取文本数据。紧接着定义数据处理逻辑先是过滤操作只有接收到的字符串是单个英文字母时才处理然后把单个英文字母映射为WordCount对象用于统计次数。然后是根据英文字母分组相同的字母会被分到同一组最后统计所有相同字母的 count 字段得到的结果就是字母出现的次数。最终的 sink 操作只是简单的把结果输出到控制台。
public class WordCountJob {public static void main(String[] args) throws Exception {// 创建Flink执行环境StreamExecutionEnvironment environment StreamExecutionEnvironment.getExecutionEnvironment();// 1. 定义数据源 从Socket读取数据environment.socketTextStream(127.0.0.1, 8888)// 2. 定义数据处理逻辑// 2.1 过滤 接收到的数据必须是英文字母.filter(e - e.length() 1 Character.isLetter(e.codePointAt(0)))// 2.2 映射 单个字符映射成WordCount对象.map(e - new WordCount(e.toUpperCase(), 1))// 2.3 分组 相同字母分为一组.keyBy(WordCount::getWord)// 2.4 分组后相同字母聚合求和.sum(count)// 3. 定义数据汇sink 这里输出到控制台.addSink(new SinkFunctionWordCount() {Overridepublic void invoke(WordCount value, Context context) throws Exception {System.err.println(value.word : value.count);}});// 执行Flink作业environment.execute();}DataAllArgsConstructorNoArgsConstructorpublic static class WordCount {public String word;public int count;}
}Flink 作业启动时就会去连接本地的8888端口如果连接不上会报错退出。所以启动作业前需要先保证8888端口开启。所以Mac系统下先打开8888端口
nc -l 8888然后启动 Flink 作业此时控制台什么也没有因为数据源没有数据Flink也就没法处理。然后我们往 8888 端口写点东西
nc -l 8888
1
a
b
c
a控制台输出
A:1
B:1
C:1
A:2分析一下结果第一次发出的“1”因为不是英文字母所以会被filter算子过滤掉。发出第一个“a”时因为符合条件所以会被后续所有算子处理最终到达sink算法输出到控制台。发出第二个“a时因为之前已经有一个a了所以sum算子求和结果是2。
尾巴
Flink作业的基本骨架包含三部分创建作业执行环境、定义数据处理逻辑、提交并启动作业。执行环境主要用来对Flink作业进行一些设置例如 故障重启策略、并行度等参数。定义数据处理逻辑是我们开发Flink作业最重要的部分首先要定义数据源告诉Flink数据从哪里来然后声明要对数据做哪些处理最终计算结果要保存到哪里等。Flink作业是懒执行的前面的这些操作都只是对Flink作业的一个声明和描述必须调用execute方法作业才会真正跑起来。
