flink使用02-从WordCount开始

相信大家在学习spark的时候接触的第一个案例肯定也是 wordCount, 本文也想通过这样一个简单的例子来讲一下一个简单的 flink 程序是什么样子的, 让大家对 flink 的代码有一个简单的了解.

一个 flink程序主要分为5个部分:

  • 1. 获取执行 Environment

environment 提供控制 job 执行的方法(例如设置并行度/容错/checkpoint 参数) 并且与外部系统做交互. flink可以做流计算也可以做批计算, 对应的也就有不同的environment , 在使用时根据不同的使用场景选择即可.

  • 2. 获取输入流 Source

    一个流式计算框架自然是少不了数据的输入, 在 streamExecutionEnvironment 的可以看到有很多种创建输入流的方式, 不过在项目中使用最多的还是使用 addSource()方法来添加不同的数据源接入

  • 3. 执行计算 Operator

    在spark中,对数据的转换计算是通过 action 算子和 transformation 算子来对 RDD 中的数据来进行操作, 而在flink中, 主要是通过 Operator来对一个流做处理, 通过一个 Operator 可以将一个流转换为另外一个流, flink中内置了很多算子来实现Operator操作.

  • 4. 输入结果 Sink

    在完成数据计算之后,就需要有一个输出的地方, 通常来讲也是通过 addSink() 方法来添加不同的数据输出目标,也可以通过 print() 来直接查看输出或者写入到csv等外部文件.

  • 5. 启动 flink,提交 job

    一个 flink 代码的启动执行, 必须通过 env.executor() 方法.这行代码主要做了以下事情:

    1. 生成StreamGraph
    2. 生成JobGraph.
    3. 生成一系列配置
    4. 将 JobGraph和配置交给 flink 集群去运行
    5. 以本地模式运行的话,可以看到启动过程,如启动能量度,web模块,jobManager,ResourceManager,taskManager等等
    6. 启动任务

以下为简单的 WordCount 代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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));
}
}
}
}
}
0%