相信大家在学习spark的时候接触的第一个案例肯定也是 wordCount, 本文也想通过这样一个简单的例子来讲一下一个简单的 flink 程序是什么样子的, 让大家对 flink 的代码有一个简单的了解.
一个 flink程序主要分为5个部分:
-
1. 获取执行 Environment
environment 提供控制 job 执行的方法(例如设置并行度/容错/checkpoint 参数) 并且与外部系统做交互. flink可以做流计算也可以做批计算, 对应的也就有不同的environment , 在使用时根据不同的使用场景选择即可. -
2. 获取输入流 Source
一个流式计算框架自然是少不了数据的输入, 在 streamExecutionEnvironment 的可以看到有很多种创建输入流的方式, 不过在项目中使用最多的还是使用 addSource()方法来添加不同的数据源接入
数据源接入方法.png
-
3. 执行计算 Operator
在spark中,对数据的转换计算是通过 action 算子和 transformation 算子来对 RDD 中的数据来进行操作, 而在flink中, 主要是通过 Operator来对一个流做处理, 通过一个 Operator 可以将一个流转换为另外一个流, flink中内置了很多算子来实现Operator操作.
-
4. 输入结果 Sink
在完成数据计算之后,就需要有一个输出的地方, 通常来讲也是通过 addSink() 方法来添加不同的数据输出目标,也可以通过 print() 来直接查看输出或者写入到csv等外部文件.
-
5. 启动 flink,提交 job
一个 flink 代码的启动执行, 必须通过 env.executor() 方法.这行代码主要做了以下事情:
- 生成StreamGraph
- 生成JobGraph.
- 生成一系列配置
- 将 JobGraph和配置交给 flink 集群去运行
- 以本地模式运行的话,可以看到启动过程,如启动能量度,web模块,jobManager,ResourceManager,taskManager等等
- 启动任务
以下为简单的 WordCount 代码
public class WordCount {
public static void main(String[] args) throws Exception {
// 获取 StreamEnv
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 获取 输入流
DataStream<String> text = env.fromElements(WordCountData.WORDS);
// 执行计算Operator
DataStream<Tuple2<String, Integer>> counts
= text.flatMap(new SplitFunction())
.keyBy(0).sum(1);
// 输出结果
counts.print();
// 启动flink程序
env.execute("WordCount Demo");
}
// *************************************************************************
// 自定义切割Function切分一行输入
// *************************************************************************
public static final class SplitFunction implements FlatMapFunction<String, Tuple2<String, Integer>>{
@Override
public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
String[] words = s.toLowerCase().split(" ");
for (String word : words) {
if (word.length() > 0){
collector.collect(new Tuple2<>(word, 1));
}
}
}
}
}
网友评论