目录
一、Flink快速上手
1.1、环境准备
1.2 创建项目
1.3 编写代码
1.3.1 批处理
1.3.2 流处理
1.4 本章总结
一、Flink快速上手
对 Flink 有了基本的了解后,接下来就要理论联系实际,真正上手写代码了。Flink 底层是 以 Java 编写的,并为开发人员同时提供了完整的 Java 和 Scala API。在本书中,代码示例将全 部用 Java 实现;而在具体项目应用中,可以根据需要选择合适语言的 API 进行开发。
在这一章,我们将会以大家最熟悉的 IntelliJ IDEA 作为开发工具,用实际项目中最常见的
Maven 作为包管理工具,在开发环境中编写一个简单的 Flink 项目,实现零基础快速上手。
1.1、环境准备
1.2 创建项目
在准备好所有的开发环境之后,我们就可以开始开发自己的第一个 Flink 程序了。首先我 们要做的,就是在 IDEA 中搭建一个 Flink 项目的骨架。我们会使用 Java 项目中常见的 Maven
来进行依赖管理。
1. 创建工程
2. 添加项目依赖
<properties> <flink.version>1.13.0</flink.version> <java.version>1.8</java.version> <scala.binary.version>2.12</scala.binary.version> <slf4j.version>1.7.30</slf4j.version>
</properties> <dependencies>
<!-- 引入 Flink 相关依赖--> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId>
21 <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients_${scala.binary.version}</artifactId> <version>${flink.version}</version>
</dependency>
<!-- 引入日志管理相关依赖--> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>${slf4j.version}</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>${slf4j.version}</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-to-slf4j</artifactId> <version>2.14.0</version>
</dependency>
</dependencies>
3、配置日志管理
在目录 src/main/resources 下添加文件:log4j.properties,内容配置如下:
log4j.rootLogger=error, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
1.3 编写代码
搭好项目框架,接下来就是我们的核心工作——往里面填充代码。我们会用一个最简单的 示例来说明 Flink 代码怎样编写:统计一段文字中,每个单词出现的频次。这就是传说中的
WordCount 程序——它是大数据领域非常经典的入门案例,地位等同于初学编程语言时的
22
Hello World。
我们的源码位于 src/main/java 目录下。首先新建一个包,命名为 com.atguigu.wc,在这个 包下我们将编写 Flink 入门的 WordCount 程序。
我们已经知道,尽管 Flink 自身的定位是流式处理引擎,但它同样拥有批处理的能力。所 以接下来,我们会针对不同的处理模式、不同的输入数据形式,分别讲述 WordCount 代码的 实现。
1.3.1 批处理
对于批处理而言,输入的应该是收集好的数据集。这里我们可以将要统计的文字,写入一 个文本文档,然后读取这个文件处理数据就可以了。
(1)在工程根目录下新建一个 input 文件夹,并在下面创建文本文件 words.txt
(2)在 words.txt 中输入一些文字,例如:
hello world
hello flink
hello java
(3)在 com.atguigu.chapter02 包下新建 Java 类 BatchWordCount,在静态 main 方法中编 写测试代码。
我们进行单词频次统计的基本思路是:先逐行读入文件数据,然后将每一行文字拆分成单 词;接着按照单词分组,统计每组数据的个数,就是对应单词的频次。
具体代码实现如下:
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector; public class BatchWordCount { public static void main(String[] args) throws Exception { // 1. 创建执行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 2. 从文件读取数据 按行读取(存储的元素就是每行的文本) DataSource<String> lineDS = env.readTextFile("input/words.txt"); // 3. 转换数据格式 FlatMapOperator<String, Tuple2<String, Long>> wordAndOne = lineDS .flatMap((String line, Collector<Tuple2<String, Long>> out) -> { String[] words = line.split(" "); for (String word : words) { out.collect(Tuple2.of(word, 1L));
23 } }) .returns(Types.TUPLE(Types.STRING, Types.LONG)); //当 Lambda 表达式
使用 Java 泛型的时候, 由于泛型擦除的存在, 需要显示的声明类型信息 // 4. 按照 word 进行分组 UnsortedGrouping<Tuple2<String, Long>> wordAndOneUG =
wordAndOne.groupBy(0); // 5. 分组内聚合统计 AggregateOperator<Tuple2<String, Long>> sum = wordAndOneUG.sum(1); // 6. 打印结果 sum.print(); }
}
flatMap方法可以映射成流,这里是将每一行的String类型转换为一个二元组,而这里的二元组使用Collector接口去收集Tuple2<String类型,Long类型>这个二元组,->后面的内容就是对映射的具体要求实现,这里是进行了word的sum求和(这里的1L是长整型1)。上述说明就是对类型转换包装成二元组的过程
需要注意的是,这种代码的实现方式,是基于 DataSet API 的,也就是我们对数据的处理 转换,是看作数据集来进行操作的。事实上 Flink 本身是流批统一的处理架构,批量的数据集
本质上也是流,没有必要用两套不同的 API 来实现。所以从 Flink 1.12 开始,官方推荐的做法 是直接使用 DataStream API,在提交任务时通过将执行模式设为 BATCH 来进行批处理:
$ bin/flink run -Dexecution.runtime-mode=BATCH BatchWordCount.jar
这样,DataSet API 就已经处于“软弃用”(soft deprecated)的状态,在实际应用中我们只 要维护一套 DataStream API 就可以了。这里只是为了方便大家理解,我们依然用 DataSet API
做了批处理的实现
1.3.2 流处理
1. 读取文件
我们同样试图读取文档 words.txt 中的数据,并统计每个单词出现的频次。这是一个“有 界流”的处理,整体思路与之前的批处理非常类似,代码模式也基本一致。
(1) 在 com.atguigu.wc 包下新建 Java 类 BoundedStreamWordCount,在静态 main 方法中 编写测试代码。具体代码实现如下:
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector; import java.util.Arrays; public class BoundedStreamWordCount { public static void main(String[] args) throws Exception { // 1. 创建流式执行环境 StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取文件
25 DataStreamSource<String> lineDSS = env.readTextFile("input/words.txt"); // 3. 转换数据格式 SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS .flatMap((String line, Collector<String> words) -> { Arrays.stream(line.split(" ")).forEach(words::collect); }) .returns(Types.STRING) .map(word -> Tuple2.of(word, 1L)) .returns(Types.TUPLE(Types.STRING, Types.LONG)); // 4. 分组 KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne .keyBy(t -> t.f0); // 5. 求和 SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS .sum(1); // 6. 打印 result.print(); // 7. 执行 env.execute(); }
}
我们可以先做个简单的解释。Flink 是一个分布式处理引擎,所以我们的程序应该也是分 布式运行的。在开发环境里,会通过多线程来模拟 Flink 集群运行。所以这里结果前的数字, 其实就指示了本地执行的不同线程,对应着 Flink 运行时不同的并行资源。这样第一个乱序的 问题也就解决了:既然是并行执行,不同线程的输出结果,自然也就无法保持输入的顺序了。
另外需要说明,这里显示的编号为 1~4,是由于运行电脑的 CPU 是 4 核,所以默认模拟 的并行线程有 4 个。这段代码不同的运行环境,得到的结果会是不同的。关于 Flink 程序并行 执行的数量,可以通过设定“并行度”(Parallelism)来进行配置,我们会在后续章节详细讲解 这些内容。
2. 读取文本流
在实际的生产环境中,真正的数据流其实是无界的,有开始却没有结束,这就要求我们需 要保持一个监听事件的状态,持续地处理捕获的数据。
为了模拟这种场景,我们就不再通过读取文件来获取数据了,而是监听数据发送端主机的 指定端口,统计发送来的文本数据中出现过的单词的个数。具体实现上,我们只要对
BoundedStreamWordCount 代码中读取数据的步骤稍做修改,就可以实现对真正无界流的处理。
(1)新建一个 Java 类 StreamWordCount,将 BoundedStreamWordCount 代码中读取文件 数据的 readTextFile 方法,替换成读取 socket 文本流的方法 socketTextStream。具体代码实现如 下:
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector; import java.util.Arrays; public class StreamWordCount { public static void main(String[] args) throws Exception { // 1. 创建流式执行环境 StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 读取文本流 DataStreamSource<String> lineDSS = env.socketTextStream("hadoop102",
7777); // 3. 转换数据格式 SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = lineDSS .flatMap((String line, Collector<String> words) -> { Arrays.stream(line.split(" ")).forEach(words::collect); }) .returns(Types.STRING)
27 .map(word -> Tuple2.of(word, 1L)) .returns(Types.TUPLE(Types.STRING, Types.LONG)); // 4. 分组 KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne .keyBy(t -> t.f0); // 5. 求和 SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS .sum(1); // 6. 打印 result.print(); // 7. 执行 env.execute(); }
}
我们会发现,输出的结果与之前读取文件的流处理非常相似。而且可以非常明显地看到, 每输入一条数据,就有一次对应的输出。具体对应关系是:输入“hello flink”,就会输出两条 统计结果(flink,1)和(hello,1);之后再输入“hello world”,同样会将 hello 和 world 的个
28
数统计输出,hello 的个数会对应增长为 2。
1.4 本章总结
本章主要实现一个 Flink 开发的入门程序——词频统计 WordCount。通过批处理和流处理 两种不同模式的实现,可以对 Flink 的 API 风格和编程方式有所熟悉,并且更加深刻地理解批 处理和流处理的不同。另外,通过读取有界数据(文件)和无界数据(socket 文本流)进行流 处理的比较,我们也可以更加直观地体会到 Flink 流处理的方式和特点。
这是我们 Flink 长征路上的第一步,是后续学习的基础。有了这番初体验,想必大家会发 现 Flink 提供了非常易用的 API,基于它进行开发并不是难事。之后我们会逐步深入展开,为 大家打开 Flink 神奇世界的大门。