0%

Flink 入门篇之 - 写个WordCount

Flink

Word Count 是数据处理框架、平台的Hello World。程序作用很简单,就是数有多少英文的单词。我们今天要使用 Flink 1.10 制作一个流版本的 Word Count。在这个流版本的 Word Count 中呢,字符串会以流的方式从输入到Flink中,并且我们会观察在处理过程中,字符统计的情况。

状态!状态!

在 Flink 的官网中,我们可以看到 Flink 社区对于 Flink 的定义是 Apache Flink® — Stateful Computations over Data Streams, 一个在数据流之上的有状态计算引擎。在正式开工之前,我们需要先对于状态有一个很初步很初步的了解。 Flink 官网上对于 State 有非常多专业的解释,我今天先初步在本文里面先谈谈简单的一些理解,感兴趣的朋友呢可以去读读 Flink 官方的解释。

在 Flink 中,任务是使用有向无环图进行描述的,每个图的节点实际上是一个操作函数或算子(Operator),而这些函数和算子在 Flink 处理单一的元素和事件时会存储数据(官网使用的是 remember 记住),因此我们可以说 Flink 的算子和函数是有状态的(Stateful)。状态在Flink中非常重要,我会在后面单独谈谈 Flink 的状态。

何为流

我们在学习程序设计的过程中,会接触到各种各样的流,文件流,网络流,Java 的流 API 等等等等。在 Flink 的世界里,数据可以形成事件流,日志记录,传感器数据,用户的行为等等都可以形成数据流。Flink 官网 中把流分为两类:有界流和无界流。

Flink 官网的解释图

简单来说,无界流有流的起点但是没有终点,数据源源不断地到达,因此我们也需要持续地处理数据。而有界流是有流的结束点的,且数据可以在全部到达后进行处理,因此我们可以把有界流理解为批处理(Batch)。与真正的批处理有所不同的是,Flink是有状态的,每个状态中的数据可能会随着数据的不断到达而发生变化。值得注意的是,由于无界流会源源不断地到达,一般我们会使用时间的特征作为无界流的处理顺序。

搭建 Maven 项目

首先创建一个 Maven 项目,并添加以下关键的依赖:

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
39
40
41
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.10.0</flink.version>
<java.version>1.8</java.version>
<scala.binary.version>2.11</scala.binary.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
</properties>

<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.15</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.7</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
<scope>runtime</scope>
</dependency>
</dependencies>

编写 Word Count Job

完成 Maven 后我们需要编写 Word Count 的 任务代码。和普通的 Java 程序一样,Flink 会找到并执行 main 方法中所定义的处理逻辑。首先我们需要在 main 方法的第一句就使用 getExecutionEnvironment() 方法获得程序运行所需要的流环境

1
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

然后我们需要向流处理的环境注册一个数据源,以源源不断地读入字符串:

1
DataStream<String> source = env.addSource(new WordCountSource()).name("SteveJobs-Word");

在获取到数据源后,我们需要对数据进行处理。首先是需要使用 flatMap 配合分词工具对输入的字符串进行切分,得到多个二元组。这些二元组的第一个元素是所得的单词,而第二个元素是 1 。为什么是 1 呢,因为我们会根据单词进行分组,每个单词都会拥有一个独立的 状态 ,每个状态都会存储当前处理过的数据信息。在这个样例中,flink 会对每个单词的状态中所有元组的第二个元素进行求和操作。由于第二个元素都是1,可知其求和的结果等于单词的个数。

1
2
3
4
DataStream<Tuple2<String,Integer>> proc = source.flatMap(new TextTokenizer())
.keyBy(0)
.sum(1)
.name("Process-Word-Count");

在 Flink 中,一切的数据都需要输出到 Sink 中。Sink 可以输出数据到数据库,到 Elasticsearch 中,到 HDFS 中,或是到 Kafka MQ 中。因此,我们需要指定 Flink 的计算结果输出位置。如在本样例中,我们会把数据输出到

1
2
proc.addSink(new WordCountSink())
.name("Word-Count-Sink");

Flink 的任务需要在本地打包,并上传到服务器加载后才可以在 Flink 集群上运行。因此我们需要在 main 方法的末尾,指定任务的名称,并调用环境的 execute 方法启动任务。

1
env.execute("Word Count");

最终,我们可以得到 WordCountJob 的写法:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class WordCountJob  {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> source = env.addSource(new WordCountSource())
.name("SteveJobs-Word");

DataStream<Tuple2<String,Integer>> proc = source.flatMap(new TextTokenizer())
.keyBy(0)
.sum(1)
.name("Process-Word-Count");

proc.addSink(new WordCountSink())
.name("Word-Count-Sink");

env.execute("Word Count");
}

}

数据的 Source 和 Sink

在这个项目中,我们先做一个有界流的试验。在 Flink 中,Source 和 Sink 是数据输入和数据输出的算子,即数据通过 Source 算子输入, 通过 Sink 算子输出。根据刚刚所介绍的,有界流实质上是指批处理,所处理的数据是有限的。因此我们的输入数据源部分我们实现了简单的遍历器数据源 (JavaDoc),直接上代码:

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
public class WordCountSource extends FromIteratorFunction<String> implements Serializable {

private static final long serialVersionUID = 0L;

public WordCountSource() {
super(new WordIterator());
}

private static class WordIterator implements Iterator<String>,Serializable{

private static final long serialVersionUID = 4L;

private int index = 0;
private int length = -1;

private WordIterator(){
length = StaticWordData.STEVE_JOBS_WORD.length;
}

@Override
public boolean hasNext() {
return (index < length);
}

@Override
public String next() {
return StaticWordData.STEVE_JOBS_WORD[index ++];
}
}

}

Sink 作为输出,我们参照了 Flink 官方 Code Walkthroughs 的写法,直接输出到日志中。 具体写法可以参考 Sink 的 JavaDoc ,以下是我的写法:

1
2
3
4
5
6
7
8
9
10
public class WordCountSink implements SinkFunction<Tuple2<String,Integer>> {

private static final long serialVersionUID = 1L;
private static final Logger logger = LoggerFactory.getLogger(WordCountSink.class);

@Override
public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
logger.info("{ Word: \""+ value.f0 + "\", Cnt:" + value.f1 +"}");
}
}

测试

在编写好代码以后,使用 Maven 的指令进行打包:

1
$ mvn clean package

我在之前的更新中介绍了如何搭建单机版 Flink , 传送门 。 我们把单机版 Flink 启动起来,然后选择 Submit New Job ,点击 Add new 通过网页版上传刚刚打包好的包。

上传好的效果图

我们可以点击 Show Plan 查看 我们在程序中定义的执行图。

执行图

点击 Submit,即可提交任务到 Flink 并执行。我们可以在 Task Manager 中,选择并查看当前执行任务的 Task Manager 的 Log。即可看到我们在 Sink 中输出的日志:

Log

可以看到,单词的统计会随着流数据的输入而不断增长。

每次更新啰里啰嗦的话

我提供了本实验的源代码,请在我的 Github 获取:ousheobin/flink-word-count

欢迎大家在上面折腾各种玩法,比如实现无界的数据源,或者把你的 word count 结果输出到 Kafka MQ。