DataStream之API
1、简介
Flink有非常灵活的分层API设计，其中的核心层就是DataStream/DataSet API。由于新版本已经实现了流批一体，DataSet API将被弃用，官方推荐统一使用DataStream API处理流数据和批数据。
DataStream（数据流）本身是Flink中一个用来表示数据集合的类（Class），我们编写的Flink代码其实就是基于这种数据类型的处理，所以这套核心API就以DataStream命名。对于批处理和流处理，我们都可以用这同一套API来实现。
DataStream在用法上有些类似于常规的Java集合，但又有所不同。我们在代码中往往并不关心集合中具体的数据，而只是用API定义出一连串的操作来处理它们；这就叫作数据流的“转换”（transformations）。
一个Flink程序，其实就是对DataStream的各种转换。具体来说，代码基本上都由以下几部分构成，如图所示：
![]()
- 获取执行环境（execution environment）
- 读取数据源（source）
- 定义基于数据的转换操作（transformations）
- 定义计算结果的输出位置（sink）
- 触发程序执行（execute）
其中，获取环境和触发执行，都可以认为是针对执行环境的操作。所以我们就从执行环境、数据源（source）、转换操作（transformation）、输出（sink）四大部分，对常用的DataStream API做基本介绍。
2、执行环境(Execution Environment)
- Flink程序可以在各种上下文环境中运行：我们可以在本地JVM中执行程序，也可以提交到远程集群上运行。
- 不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前Flink的运行环境，从而建立起与Flink框架之间的联系。只有获取了环境上下文信息，才能将具体的任务调度到不同的TaskManager执行。
2.1 创建执行环境
编写Flink程序的第 一 步，就是创建执行环境。我们要获取的执行环境，是StreamExecutionEnvironment类的对象，这是所有Flink程序的基础。在代码中创建执行环境的 方式，就是调用这个类的静态方法，具体有以下三种。
getExecutionEnvironment。最简单的方式，就是直接调用getExecutionEnvironment方法。它会根据当前运行的上下文直接得到正确的结果：如果程序是独立运行的，就返回一个本地执行环境；如果是创建了jar包，然后从命令行调用它并提交到集群执行，那么就返回集群的执行环境。也就是说，这个方法会根据当前运行的方式，自行决定该返回什么样的运行环境。这种“智能”的方式不需要我们额外做判断，用起来简单高效，是最常用的一种创建执行环境的方式。
1
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
createLocalEnvironment。这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的CPU核心数。
1
2StreamExecutionEnvironment localEnv =
StreamExecutionEnvironment.createLocalEnvironment();createRemoteEnvironment。这个方法返回集群执行环境。需要在调用时指定JobManager的主机名和端口号，并指定要在集群中运行的Jar包。在获取到程序执行环境后，我们还可以对执行环境进行灵活的设置。比如可以全局设置程序的并行度、禁用算子链，还可以定义程序的时间语义、配置容错机制。
1
2
3
4
5
6StreamExecutionEnvironment remoteEnv = StreamExecutionEnvironment
.createRemoteEnvironment(
"host", // JobManager 主机名
1234, // JobManager 进程端口号
"path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
);
2.2 执行模式(Execution Mode)
上节中我们获取到的执行环境，是一个StreamExecutionEnvironment，顾名思义它应该是做流处理的。那对于批处理，又应该怎么获取执行环境呢？在之前的Flink版本中，批处理的执行环境与流处理类似，是调用类 ExecutionEnvironment的静态方法，返回它的对象：
1
2
3
4
5// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 流处理环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();基于ExecutionEnvironment读入数据创建的数据集合，就是DataSet；对应的调用的一整套转换方法，就是DataSet API。而从1.12.0版本起，Flink实现了API上的流批统一。DataStream API新增了一个重要特性：可以支持不同的“执行模式”（execution mode），通过简单的设置就可以让一段Flink程序在流处理和批处理之间切换。这样一来，DataSet API也就没有存在的必要了。
流执行模式（STREAMING）。这是DataStream API最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是STREAMING执行模式。
批执行模式（BATCH）。专门用于批处理的执行模式，这种模式下，Flink处理作业的方式类似于MapReduce框架。 对于不会持续计算的有界数据，我们用这种模式处理会更方便。主要有两种方式：
通过命令行配置，在提交作业时，增加execution.runtime-mode参数，指定值为 BATCH：
1
bin/flink run -Dexecution.runtime-mode=BATCH ...
通过代码配置，直接基于执行环境调用setRuntimeMode方法，传入BATCH模式：
1
2
3StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
建议：
不要在代码中配置，而是使用命令行。这同设置并行度是类似的：在提交作业时指定参数可以更加灵活，同一段应用程序写好之后，既可以用于批处理也可以用于流处理。而在代码中硬编码（hard code）的方式可扩展性比较差，一般都不推荐。
用BATCH模式处理批量数据，用STREAMING模式处理流式数据。因为数据有界的时候，直接输出结果会更加高效；而当数据无界的时候，我们没得选择——只有STREAMING模式才能处理持续的数据流。
自动模式（AUTOMATIC）。在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。
2.3 触发程序执行
有了执行环境，我们就可以构建程序的处理流程了：基于环境读取数据源，进而进行各种转换操作，最后输出结果到外部系统。
需要注意的是，写完输出（sink）操作并不代表程序已经结束。因为当main()方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据——因为数据可能还没来。Flink是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为“延迟执行”或“懒执行”（lazy execution）。
所以我们需要显式地调用执行环境的execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后返回一个执行结果（JobExecutionResult）。
1
env.execute();
3、源算子(Source)
创建环境之后，就可以构建数据处理的业务逻辑了，如图所示，本节将主要讲解Flink的源算子（Source）。想要处理数据，先得有数据，所以首要任务就是把数据读进来。
![]()
Flink可以从各种来源获取数据，然后构建DataStream进行转换处理。一般将数据的输入来源称为数据源(data source)，而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。Flink代码中通用的添加source的方式，是调用执行环境的addSource()方法：
1
DataStream<String> stream = env.addSource(...);
方法传入一个对象参数，需要实现SourceFunction接口；返回DataStreamSource。这里的 DataStreamSource类继承自SingleOutputStreamOperator类，又进一步继承自DataStream。所以很明显，读取数据的source操作是一个算子，得到的是一个数据流（DataStream）。
![1662212401413]()
3.1 准备工作
为了更好地理解，我们先构建一个实际应用场景。比如网站的访问操作，可以抽象成一个三元组（用户名，用户访问的url，用户访问url的时间戳），所以在这里，我们可以创建一个类Event，将用户行为包装成它的一个对象。Event包含了以下一些字段，如表所示：
![]()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20public class Event {
public String user;
public String url;
public Long timestamp;
public Event() {
}
public Event(String user, String url, Long timestamp) {
this.user = user;
this.url = url;
this.timestamp = timestamp;
}
public String toString() {
return "Event{" +
"user='" + user + '\'' +
", url='" + url + '\'' +
", timestamp=" + new Timestamp(timestamp) +
'}';
}
}这里需要注意，我们定义的Event，有这样几个特点：
- 类是公有（public）的。
- 有一个无参的构造方法。
- 所有属性都是公有（public）的。
- 所有属性的类型都是可以序列化的。
Flink会把这样的类作为一种特殊的POJO数据类型来对待，方便数据的解析和序列化。 另外我们在类中还重写了toString方法，主要是为了测试输出显示更清晰。
3.2 从集合中读取数据
最简单的读取数据的方式，就是在代码中直接创建一个Java集合，然后调用执行环境的fromCollection方法进行读取。这相当于将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用，一般用于测试。
1
2
3
4
5
6
7
8
9
10
11
12
13public class TestSource {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
ArrayList<Event> clicks = new ArrayList<>();
clicks.add(new Event("Mary", "./home", 1000L));
clicks.add(new Event("Bob", "./cart", 2000L));
DataStream<Event> stream = env.fromCollection(clicks);
stream.print();
env.execute();
}
}我们也可以不构建集合，直接将元素列举出来，调用fromElements方法进行读取数据：
1
2
3
4
5
6
7
8
9
10
11
12
13public class TestSource {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
stream.print();
env.execute();
}
}
3.3 从文件读取数据
真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。
1
DataStream<String> stream = env.readTextFile("clicks.csv");
参数可以是目录，也可以是文件；
路径可以是相对路径，也可以是绝对路径；
相对路径是从系统属性user.dir获取路径：idea下是project的根目录，standalone模式下是集群节点根目录；
也可以从hdfs目录下读取, 使用路径hdfs://…，由于Flink没有提供hadoop相关依赖, 需要pom中添加相关依赖：
1
2
3
4
5
6<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.5</version>
<scope>provided</scope>
</dependency>
3.4 从Socket读取数据
不论从集合还是文件，我们读取的其实都是有界数据。在流处理的场景中，数据往往是无界的。这时又从哪里读取呢？一个简单的方式，就是我们之前用到的读取socket文本流。这种方式由于吞吐量小、稳定性较差，一般也是用于测试。
1
DataStream<String> stream = env.socketTextStream("localhost", 7777);
3.5 从Kafka读取数据
Kafka作为分布式消息传输队列，是一个高吞吐、易于扩展的消息系统。而消息队列的传输方式，恰恰和流处理是完全一致的。所以可以说Kafka和Flink天生一对，是当前处理流式数据的双子星。在如今的实时流处理应用中，由Kafka进行数据的收集和传输，Flink进行分析计算，这样的架构已经成为众多企业的首选，如图所示。
![]()
略微遗憾的是，与Kafka的连接比较复杂，Flink内部并没有提供预实现的方法。所以我们只能采用通用的addSource方式、实现一个SourceFunction了。好在Kafka与Flink确实是非常契合，所以Flink官方提供了连接工具flink-connector-kafka， 直接帮我们实现了一个消费者FlinkKafkaConsumer，它就是用来读取Kafka数据的SourceFunction。
所以想要以Kafka作为数据源获取数据，我们只需要引入Kafka连接器的依赖。Flink官方提供的是一个通用的Kafka连接器，它会自动跟踪最新版本的Kafka客户端。目前最新版本只支持0.10.0版本以上的Kafka，使用时可以根据自己安装的Kafka版本选定连接器的依赖版本。这里我们需要导入的依赖如下。
1
2
3
4
5<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>然后调用env.addSource()，传入FlinkKafkaConsumer的对象实例就可以了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23public class SourceKafkaTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "consumer-group");
properties.setProperty("key.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("auto.offset.reset", "latest");
DataStreamSource<String> stream = env.addSource(new
FlinkKafkaConsumer<String>(
"clicks",
new SimpleStringSchema(),
properties
));
stream.print("Kafka");
env.execute();
}
}- 第一个参数topic，定义了从哪些主题中读取数据。可以是一个topic，也可以是topic列表，还可以是匹配所有想要读取的topic的正则表达式。当从多个topic中读取数据时，Kafka连接器将会处理所有topic的分区，将这些分区的数据放到一条流中去。
- 第二个参数是一个DeserializationSchema或者KeyedDeserializationSchema。Kafka消息被存储为原始的字节数据，所以需要反序列化成Java或者Scala对象。上面代码中使用的SimpleStringSchema，是一个内置的DeserializationSchema，它只是将字节数组简单地反序列化成字符串。DeserializationSchema 和KeyedDeserializationSchema是公共接口，所以我们也可以自定义反序列化逻辑。
- 第三个参数是一个Properties对象，设置了Kafka客户端的一些属性。
3.6 自定义Source
大多数情况下，前面的数据源已经能够满足需要。但是凡事总有例外，如果遇到特殊情况，我们想要读取的数据源来自某个外部系统，而flink既没有预实现的方法、也没有提供连接器，那就只好自定义实现SourceFunction了。接下来我们创建一个自定义的数据源，实现SourceFunction接口。主要重写两个关键方法：run()和cancel()。
- run()方法：使用运行时上下文对象（SourceContext）向下游发送数据；
- cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。
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
27public class ClickSource implements SourceFunction<Event> {
// 声明一个布尔变量，作为控制数据生成的标识位
private Boolean running = true;
public void run(SourceContext<Event> ctx) throws Exception {
Random random = new Random();
// 在指定的数据集中随机选取数据
String[] users = {"Mary", "Alice", "Bob", "Cary"};
String[] urls = {"./home", "./cart", "./fav", "./prod?id=1",
"./prod?id=2"};
while (running) {
ctx.collect(new Event(
users[random.nextInt(users.length)],
urls[random.nextInt(urls.length)],
Calendar.getInstance().getTimeInMillis()
));
// 隔 1 秒生成一个点击事件，方便观测
Thread.sleep(1000);
}
}
public void cancel() {
running = false;
}
}使用自定义数据源：
1
2
3
4
5
6
7
8
9
10
11public class SourceCustom {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
//有了自定义的 source function，调用 addSource 方法
DataStreamSource<Event> stream = env.addSource(new ClickSource());
stream.print("SourceCustom");
env.execute();
}
}![]()
这里要注意的是SourceFunction接口定义的数据源，并行度只能设置为1，如果数据源设置为大于1的并行度，则会抛出异常。如下程序所示：
1
2
3
4
5
6
7
8public class SourceCustom {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new ClickSource()).setParallelism(2).print();
env.execute();
}
}![]()
所以如果我们想要自定义并行的数据源的话，需要使用ParallelSourceFunction，示例程序如下：
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
26public class ParallelSourceExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.addSource(new CustomSource()).setParallelism(2).print();
env.execute();
}
public static class CustomSource implements ParallelSourceFunction<Integer> {
private boolean running = true;
private Random random = new Random();
public void run(SourceContext<Integer> sourceContext) throws Exception {
while (running) {
sourceContext.collect(random.nextInt());
}
}
public void cancel() {
running = false;
}
}
}![]()
3.7 Flink支持的数据类型
为了方便地处理数据，Flink有自己一整套类型系统。Flink使用“类型信息” （TypeInformation）来统一表示数据类型。TypeInformation类是Flink中所有类型描述符的基类。它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。
简单来说，对于常见的Java和Scala数据类型，Flink都是支持的。Flink在内部，Flink对支持不同的类型进行了划分，这些类型可以在Types工具类中找到：
- 基本类型。所有Java基本类型及其包装类，再加上Void、String、Date、BigDecimal 和BigInteger。
- 数组类型。包括基本类型数组（PRIMITIVE_ARRAY）和对象数组(OBJECT_ARRAY)。
- 复合数据类型。
- Java元组类型（TUPLE）：这是Flink内置的元组类型，是Java API的一部分。最多25个字段，也就是从Tuple0~Tuple25，不支持空字段。
- Scala样例类及Scala元组：不支持空字段。
- 行类型（ROW）：可以认为是具有任意个字段的元组，并支持空字段。
- POJO：Flink自定义的类似于Java bean模式的类。
- 辅助类型。Option、Either、List、Map等。
- 泛型类型（GENERIC）。
Flink支持所有的Java类和Scala类。不过如果没有按照上面POJO类型的要求来定义，就会被Flink当作泛型类来处理。Flink会把泛型类型当作黑盒，无法获取它们内部的属性；它们也不是由Flink本身序列化的，而是由Kryo序列化的。
在这些类型中，元组类型和POJO类型最为灵活，因为它们支持创建复杂类型。而相比之下，POJO还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。所以，在项目实践中，往往会将流处理程序中的元素类型定为Flink的POJO类型。Flink对POJO类型的要求如下：
- 类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；
- 类有一个公共的无参构造方法；
- 类中的所有字段是public且非final的；或者有一个公共的getter和setter方法，这些方法需要符合Java bean的命名规范。
所以我们看到，之前的UserBehavior，就是我们创建的符合Flink POJO定义的数据类型。
类型提示（Type Hints）
Flink还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息， 从而获得对应的序列化器和反序列化器。但是，由于Java中泛型擦除的存在，在某些特殊情况下（比如Lambda表达式中），自动提取的信息是不够精细的——只告诉Flink当前的元素由 “船头、船身、船尾”构成，根本无法重建出“大船”的模样；这时就需要显式地提供类型信息，才能使应用程序正常工作或提高其性能。为了解决这类问题，Java API 提供了专门的“类型提示”（type hints）。
回忆一下之前的word count流处理程序，我们在将String类型的每个词转换成（word， count）二元组后，就明确地用returns指定了返回的类型。因为对于map里传入的Lambda表达式，系统只能推断出返回的是Tuple2类型，而无法得到Tuple2。只有显式地告诉系统当前的返回类型，才能正确地解析出完整数据。
1
2.map(word -> Tuple2.of(word, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));这是一种比较简单的场景，二元组的两个元素都是基本数据类型。那如果元组中的一个元素又有泛型，该怎么处理呢？Flink专门提供了TypeHint类，它可以捕获泛型的类型信息，并且一直记录下来，为运行时提供足够的信息。我们同样可以通过.returns()方法，明确地指定转换之后的DataStream里元素的类型。
1
returns(new TypeHint<Tuple2<Integer, SomeType>>(){})
4、转换算子(Transformation)
数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个DataStream转换为新的DataStream，如图所示。一个Flink程序的核心，其实就是所有的转换操作，它们决定了处理的业务逻辑。
![]()
我们可以针对一条流进行转换处理，也可以进行分流、合流等多流转换操作，从而组合成复杂的数据流拓扑。
4.1 基本转换算子
4.1.1 映射(map)
map是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素，如图所示。
![]()
我们只需要基于DataStrema调用map()方法就可以进行转换处理。方法需要传入的参数是接口MapFunction的实现；返回值类型还是DataStream，不过泛型（流中的元素类型）可能改变。下面的代码用不同的方式，实现了提取Event中的user字段的功能。
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
28public class TransMapTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// 传入匿名类，实现 MapFunction
stream.map(new MapFunction<Event, String>() {
public String map(Event e) throws Exception {
return e.user;
}
});
// 传入 MapFunction 的实现类
stream.map(new UserExtractor()).print();
env.execute();
}
public static class UserExtractor implements MapFunction<Event, String> {
public String map(Event e) throws Exception {
return e.user;
}
}
}![]()
MapFunction实现类的泛型类型，与输入数据类型和输出数据的类型有关。在实现MapFunction接口的时候，需要指定两个泛型，分别是输入事件和输出事件的类型，还 需要重写一个map()方法，定义从一个输入事件转换为另一个输出事件的具体逻辑。
通过查看Flink源码可以发现，基于DataStream调用map方法，返回的其实是一个SingleOutputStreamOperator。
1
public <R> SingleOutputStreamOperator<R> map(MapFunction<T, R> mapper){}
- 这表示map是一个用户可以自定义的转换（transformation）算子，它作用于一条数据流上，转换处理的结果是一个确定的输出类型。当然，SingleOutputStreamOperator类本身也继承自DataStream类，所以说map是将一个DataStream转换成另一个DataStream是完全正确的。
4.1.2 过滤(filter)
filter转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤条件，对于每一个流内元素进行判断，若为true则元素正常输出，若为false则元素被过滤掉，如图所示。
![]()
进行filter转换之后的新数据流的数据类型与原数据流是相同的。filter转换需要传入的参数需要实现FilterFunction接口，而FilterFunction内要实现filter()方法，就相当于一个返回布尔类型的条件表达式。下面的代码会将数据流中用户Mary的浏览行为过滤出来。
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
28public class TransFilterTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// 传入匿名类实现 FilterFunction
stream.filter(new FilterFunction<Event>() {
public boolean filter(Event e) throws Exception {
return e.user.equals("Mary");
}
});
// 传入 FilterFunction 实现类
stream.filter(new UserFilter()).print();
env.execute();
}
public static class UserFilter implements FilterFunction<Event> {
public boolean filter(Event e) throws Exception {
return e.user.equals("Mary");
}
}
}![]()
4.1.3 扁平映射(flatMap)
flatMap操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个一个的个体使用。消费一个元素，可以产生0到多个元素。flatMap可以认为是“扁平化”（flatten） 和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分后的元素做转换处理，如图所示。此前WordCount程序的第一步分词操作，就用到了flatMap。
![]()
同map一样，flatMap也可以使用Lambda表达式或者FlatMapFunction接口实现类的方式来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。
flatMap操作会应用在每一个输入事件上面，FlatMapFunction接口中定义了flatMap方法， 用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回0个、1个或多个结果数据。因此flatMap并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来 指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调用，也可以不调用。所以flatMap方法也可以实现map方法和filter方法的功能，当返回结果是0个的时候，就相当于对数据进行了过滤，当返回结果是1个的时候，相当于对数据进行了简单的转换操作。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25public class TransFlatmapTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
stream.flatMap(new MyFlatMap()).print();
env.execute();
}
public static class MyFlatMap implements FlatMapFunction<Event, String> {
public void flatMap(Event value, Collector<String> out) throws Exception {
if (value.user.equals("Mary")) {
out.collect(value.user);
} else if (value.user.equals("Bob")) {
out.collect(value.user);
out.collect(value.url);
}
}
}
}![]()
4.2 聚合算子(Aggregation)
- 直观上看，基本转换算子确实是在“转换”——因为它们都是基于当前数据，去做了处理和输出。而在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的信息。比如之前word count程序中，要对每个词出现的频次进行叠加统计。这种操作，计算的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并 ——这就是所谓的“聚合”（Aggregation），也对应着MapReduce中的reduce操作。
4.2.1 按键分区(keyBy)
对于Flink而言，DataStream是没有直接进行聚合的API的。因为我们对海量数据做聚合肯定要进行分区并行处理，这样才能提高效率。所以在Flink中，要做聚合，需要先进行分区；这个操作就是通过keyBy来完成的。
keyBy是聚合前必须要用到的一个算子。keyBy通过指定键（key），可以将一条流从逻辑上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应着任务槽（task slot）。
基于不同的key，流中的数据将被分配到不同的分区中去，如图所示；这样一来，所有具有相同的key的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个slot中进行处理了。
![]()
在内部，是通过计算key的哈希值（hash code），对分区数进行取模运算来实现的。所以这里key如果是POJO的话，必须要重写hashCode()方法。
keyBy()方法需要传入一个参数，这个参数指定了一个或一组key。有很多不同的方法来指定key：比如对于Tuple数据类型，可以指定字段的位置或者多个位置的组合；对于POJO类型，可以指定字段的名称（String）；另外，还可以传入Lambda表达式或者实现一个键选择器 （KeySelector），用于说明从数据中提取key的逻辑。我们可以以id作为key做一个分区操作，代码实现如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21public class TransKeyByTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// 使用 Lambda 表达式
KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user);
// 使用匿名类实现 KeySelector
KeyedStream<Event, String> keyedStream1 = stream.keyBy(new KeySelector<Event, String>() {
public String getKey(Event e) throws Exception {
return e.user;
}
});
env.execute();
}
}需要注意的是，keyBy得到的结果将不再是DataStream，而是会将DataStream转换为KeyedStream。KeyedStream可以认为是“分区流”或者“键控流”，它是对DataStream按照key的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定key的类型。
KeyedStream也继承自DataStream，所以基于它的操作也都归属于DataStream API。但它跟之前的转换操作得到的SingleOutputStreamOperator不同，只是一个流的分区操作，并不是一个转换算子。KeyedStream 是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如 sum，reduce）；而且它可以将当前算子任务的状态（state）也按照key进行划分、限定为仅对当前key有效。
4.2.2 简单聚合
有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink为我们内置实现了一些最基本、最简单的聚合API，主要有以下几种：
- sum()：在输入流上，对指定的字段做叠加求和的操作。
- min()：在输入流上，对指定的字段求最小值。
- max()：在输入流上，对指定的字段求最大值。
- minBy()：与min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计算指定字段的最小值，其他字段会保留最初第一个数据的值；而minBy()则会返回包含字段最小值的整条数据。
- maxBy()：与max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。
简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数；但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字段的方式有两种：指定位置，和指定名称。
对于元组类型的数据，同样也可以使用这两种方式来指定字段。需要注意的是，元组中字段的名称，是以f0、f1、f2、…来命名的。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24public class TransTupleAggreationTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Tuple2<String, Integer>> stream = env.fromElements(
Tuple2.of("a", 1),
Tuple2.of("a", 3),
Tuple2.of("b", 3),
Tuple2.of("b", 4)
);
stream.keyBy(r -> r.f0).sum(1).print();
stream.keyBy(r -> r.f0).sum("f1").print();
stream.keyBy(r -> r.f0).max(1).print();
stream.keyBy(r -> r.f0).max("f1").print();
stream.keyBy(r -> r.f0).min(1).print();
stream.keyBy(r -> r.f0).min("f1").print();
stream.keyBy(r -> r.f0).maxBy(1).print();
stream.keyBy(r -> r.f0).maxBy("f1").print();
stream.keyBy(r -> r.f0).minBy(1).print();
stream.keyBy(r -> r.f0).minBy("f1").print();
env.execute();
}
}而如果数据流的类型是 POJO 类，那么就只能通过字段名称来指定，不能通过位置来指定了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14public class TransPojoAggregationTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Bob", "./cart1", 3000L)
);
stream.keyBy(e -> e.user).max("timestamp").print(); // 指定字段名称
env.execute();
}
}![]()
简单聚合算子返回的，同样是一个SingleOutputStreamOperator，也就是从KeyedStream又转换成了常规的 DataStream。所以可以这样理解：keyBy和聚合是成对出现的，先分区、后聚合，得到的依然是一个DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不变。
一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。 所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子，应该只用在含有有限个key的数据流上。
4.2.3 归约聚合(reduce)
与简单聚合类似，reduce操作也会将KeyedStream转换为DataStream。它不会改变流的元素数据类型，所以输出类型和输入类型是一样的。调用KeyedStream的reduce方法时，需要传入一个参数，实现ReduceFunction接口。接口在源码中的定义如下：
1
2
3public interface ReduceFunction<T> extends Function, Serializable {
T reduce(T value1, T value2) throws Exception;
}ReduceFunction接口里需要实现reduce()方法，这个方法接收两个输入事件，经过转换处理之后输出一个相同类型的事件；所以，对于一组数据，我们可以先取两个进行合并，然后再将合并的结果看作一个数据、再跟后面的数据合并，最终会将它“简化”成唯一的一个数据， 这也就是reduce“归约”的含义。在流处理的底层实现过程中，实际上是将中间“合并的结果” 作为任务的一个状态保存起来的；之后每来一个新的数据，就和之前的聚合状态进一步做归约。
我们将数据流按照用户id进行分区，然后用一个reduce算子实现sum的功能，统计每个用户访问的频次；进而将所有统计结果分到一组，用另一个reduce算子实现maxBy的功能，记录所有用户中访问频次最高的那个，也就是当前访问量最大的用户是谁。
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
38public class TransReduceTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 这里的 ClickSource()使用了之前自定义数据源小节中的 ClickSource()
env.addSource(new ClickSource())
// 将 Event 数据类型转换成元组类型
.map(new MapFunction<Event, Tuple2<String, Long>>() {
public Tuple2<String, Long> map(Event e) throws Exception {
return Tuple2.of(e.user, 1L);
}
})
.keyBy(r -> r.f0)
// 使用用户名来进行分流
.reduce(new ReduceFunction<Tuple2<String, Long>>() {
public Tuple2<String, Long> reduce(Tuple2<String, Long> value1,
Tuple2<String, Long> value2) throws Exception {
// 每到一条数据，用户 pv 的统计值加 1
return Tuple2.of(value1.f0, value1.f1 + value2.f1);
}
})
.keyBy(r -> true)
// 为每一条数据分配同一个 key，将聚合结果发送到一条流中去
.reduce(new ReduceFunction<Tuple2<String, Long>>() {
public Tuple2<String, Long> reduce(Tuple2<String, Long> value1,
Tuple2<String, Long> value2) throws Exception {
// 将累加器更新为当前最大的 pv 统计值，然后向下游发送累加器的值
return value1.f1 > value2.f1 ? value1 : value2;
}
})
.print();
env.execute();
}
}![]()
reduce同简单聚合算子一样，也要针对每一个key保存状态。因为状态不会清空，所以我们需要将reduce算子作用在一个有限key的流上。
4.3 用户自定义函数(UDF)
4.3.1 函数类(Function Classes)
对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口，来完成处理逻辑的定义。Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等。下面例子实现了FilterFunction接口，用来筛选url中包含“home”的事件：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21public class TransFunctionUDFTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> clicks = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
DataStream<Event> stream = clicks.filter(new FlinkFilter());
stream.print();
env.execute();
}
public static class FlinkFilter implements FilterFunction<Event> {
public boolean filter(Event value) throws Exception {
return value.url.contains("home");
}
}
}当然还可以通过匿名类来实现FilterFunction接口：
1
2
3
4
5
6DataStream<String> stream = clicks.filter(new FilterFunction<Event>() {
public boolean filter(Event value) throws Exception {
return value.url.contains("home");
}
});为了类可以更加通用，我们还可以将用于过滤的关键字”home”抽象出来作为类的属性，调用构造方法时传进去。
1
2
3
4
5
6
7
8
9
10
11
12
13
14DataStream<Event> stream = clicks.filter(new KeyWordFilter("home"));
public static class KeyWordFilter implements FilterFunction<Event> {
private String keyWord;
KeyWordFilter(String keyWord) {
this.keyWord = keyWord;
}
public boolean filter(Event value) throws Exception {
return value.url.contains(this.keyWord);
}
}
4.3.2 匿名函数(Lambda)
匿名函数（Lambda 表达式）是Java 8引入的新特性，方便我们更加快速清晰地写代码。Lambda表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。Flink的所有算子都可以使用Lambda表达式的方式来进行编码，但是，当Lambda表达式使用Java的泛型时，我们需要显式的声明类型信息。
下例演示了如何使用Lambda表达式来实现一个简单的map()函数，我们使用Lambda表达式来计算输入的平方。在这里，我们不需要声明map()函数的输入i和输出参数的数据类型，因为Java编译器会对它们做出类型推断。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16public class TransFunctionLambdaTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> clicks = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
//map 函数使用 Lambda 表达式，返回简单类型，不需要进行类型声明
DataStream<String> stream1 = clicks.map(event -> event.url);
stream1.print();
env.execute();
}
}- 由于OUT是String类型而不是泛型，所以Flink可以从函数签名OUT map(IN value)的实现中自动提取出结果的类型信息。
但是对于像flatMap()这样的函数，它的函数签名void flatMap(IN value, Collector out) 被Java编译器编译成了 void flatMap(IN value, Collector out)，也就是说将Collector的泛型信息擦除掉了。这样Flink就无法自动推断输出的类型信息了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16public class TransFlatmapTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// flatMap 使用 Lambda 表达式，抛出异常
stream.flatMap((event, out) -> {
out.collect(event.url);
}).print();
env.execute();
}
}![]()
在这种情况下，我们需要显式地指定类型信息，否则输出将被视为Object类型，这会导致低效的序列化。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17public class TransFlatmapTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// flatMap 使用 Lambda 表达式，必须通过 returns 明确声明返回类型
stream.flatMap((Event event, Collector<String>
out) -> {
out.collect(event.url);
}).returns(Types.STRING).print();
env.execute();
}
}当使用map()函数返回Flink自定义的元组类型时也会发生类似的问题。下例中的函数签名Tuple2 map(Event value)被类型擦除为Tuple2 map(Event value)。
1
2
3
4//使用 map 函数也会出现类似问题，以下代码会报错
DataStream<Tuple2<String, Long>> stream3 = clicks
.map( event -> Tuple2.of(event.user, 1L) );
stream3.print();一般来说，这个问题可以通过多种方式解决：
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
36public class ReturnTypeResolve {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> clicks = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// 想要转换成二元组类型，需要进行以下处理
// 1) 使用显式的 ".returns(...)"
DataStream<Tuple2<String, Long>> stream3 = clicks
.map(event -> Tuple2.of(event.user, 1L))
.returns(Types.TUPLE(Types.STRING, Types.LONG));
stream3.print();
// 2) 使用类来替代 Lambda 表达式
clicks.map(new MyTuple2Mapper()).print();
// 3) 使用匿名类来代替 Lambda 表达式
clicks.map(new MapFunction<Event, Tuple2<String, Long>>() {
public Tuple2<String, Long> map(Event value) throws Exception {
return Tuple2.of(value.user, 1L);
}
}).print();
env.execute();
}
// 自定义 MapFunction 的实现类
public static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String,
Long>> {
public Tuple2<String, Long> map(Event value) throws Exception {
return Tuple2.of(value.user, 1L);
}
}
}
4.3.3 富函数类(Rich Function Classes)
“富函数类”也是DataStream之API提供的一个函数类的接口，所有的Flink函数类都有其Rich版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction等。
既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
Rich Function有生命周期的概念。典型的生命周期方法有：
- open()方法，是Rich Function的初始化方法，也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如map()或者filter()方法被调用之前，open()会首先被调 用。所以像文件IO的创建，数据库连接的创建，配置文件的读取等等这样一次性的工作，都适合在open()方法中完成。
- close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一些清理工作。
需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的，实际工作方法，例如RichMapFunction中的map()，在每条数据到来后都会触发一次调用。
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
36public class RichFunctionTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
DataStreamSource<Event> clicks = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Alice", "./prod?id=1", 5 * 1000L),
new Event("Cary", "./home", 60 * 1000L)
);
// 将点击事件转换成长整型的时间戳输出
clicks.map(new RichMapFunction<Event, Long>() {
public void open(Configuration parameters) throws Exception {
super.open(parameters);
System.out.println(" 索 引 为 " +
getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始");
}
public Long map(Event value) throws Exception {
return value.timestamp;
}
public void close() throws Exception {
super.close();
System.out.println(" 索 引 为 " +
getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束");
}
})
.print();
env.execute();
}
}![]()
一个常见的应用场景就是，如果我们希望连接到一个外部数据库进行读写操作，那么将连接操作放在map()中显然不是个好选择——因为每来一条数据就会重新连接一次数据库；所以我们可以在open()中建立连接，在 map()中读写数据，而在close()中关闭连接。所以我们推荐的最佳实践如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17public class MyFlatMap extends RichFlatMapFunction<IN, OUT> {
public void open(Configuration configuration) {
// 做一些初始化工作
// 例如建立一个和 MySQL 的连接
}
public void flatMap(IN in, Collector<OUT> out) {
// 对数据库进行读写
}
public void close() {
// 清理工作，关闭和 MySQL 数据库的连接。
}
}另外，富函数类提供了getRuntimeContext()方法，可以获取到运行时上下文的一些信息，例如程序执行的并行度，任务名称，以及状态（state）。这使得我们可以大大扩展程序的功能，特别是对于状态的操作，使得Flink中的算子具备了处理复杂业务的能力。
4.4 物理分区(Physical Partitioning)
顾名思义，“分区”（partitioning）操作就是要将数据进行重新分布，传递到不同的流分区去进行下一步处理。其实我们对分区操作并不陌生，前面介绍聚合算子时，已经提到了keyBy，它就是一种按照键的哈希值来进行重新分区的操作。只不过这种分区操作只能保证把数据按key“分开”，至于分得均不均匀、每个key的数据具体会分到哪一区去，这些是完全无从控制的——所以我们有时也说，keyBy是一种逻辑分区（logical partitioning）操作。
为了同keyBy相区别，我们把这些操作统称为“物理分区” 操作。物理分区与keyBy另一大区别在于，keyBy之后得到的是一个KeyedStream，而物理分区之后结果仍是DataStream，且流中元素数据类型保持不变。从这一点也可以看出，分区算子 并不对数据进行转换处理，只是定义了数据的传输方式。
常见的物理分区策略有随机分配（Random）、轮询分配（Round-Robin）、重缩放（Rescale）和广播（Broadcast）。
4.4.1 随机分区(shuffle)
最简单的重分区方式就是直接“洗牌”。通过调用DataStream的.shuffle()方法，将数据随机地分配到下游算子的并行任务中去。
随机分区服从均匀分布（uniform distribution），所以可以把流中的数据随机打乱，均匀地传递到下游任务分区，如图所示。因为是完全随机的，所以对于同样的输入数据，每次执行得到的结果也不会相同。
![]()
经过随机分区之后，得到的依然是一个DataStream。我们可以做个简单测试：将数据读入之后直接打印到控制台，将输出的并行度设置为4，中间经历一次shuffle。执行多次，观察结果是否相同。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22public class ShuffleTest {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 读取数据源，并行度为 1
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Mary", "./home", 1000L),
new Event("Mary", "./home", 1000L),
new Event("Mary", "./home", 1000L),
new Event("Mary", "./home", 1000L),
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L)
);
// 经洗牌后打印输出，并行度为 4
stream.shuffle().print("shuffle").setParallelism(4);
env.execute();
}
}![]()
从结果可以看出由于数据量比较骚导致不是严格的均匀分布，但数据量足够多时会服从均匀分布。
4.1.2 轮询分区(Round-Robin,默认)
轮询也是一种常见的重分区方式。简单来说就是“发牌”，按照先后顺序将数据做依次分发，如图所示。通过调用DataStream的.rebalance()方法，就可以实现轮询重分区。rebalance使用的是Round-Robin负载均衡算法，可以将输入流数据平均分配到下游的并行任务中去。Round-Robin算法用在了很多地方，例如Kafka和Nginx。
![]()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22public class ShuffleTest {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 读取数据源，并行度为 1
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 1200L),
new Event("Mary", "./home", 1300L),
new Event("Mary", "./home", 1400L),
new Event("Mary", "./home", 1500L),
new Event("Mary", "./home", 1600L),
new Event("Mary", "./home", 1700L),
new Event("Bob", "./cart", 1800L)
);
// 经轮询重分区后打印输出，并行度为 4
stream.rebalance().print("shuffle").setParallelism(4);
env.execute();
}
}![]()
4.1.3 重缩放分区(rescale)
重缩放分区和轮询分区非常相似。当调用rescale()方法时，其实底层也是使用Round-Robin算法进行轮询，但是只会将数据轮询发送到下游并行任务的一部分中，如图所示。也就是说，“发牌人”如果有多个，那么rebalance的方式是每个发牌人都面向所有人发牌；而rescale的做法是分成小团体，发牌人只给自己团体内的所有人轮流发牌。
![]()
当下游任务（数据接收方）的数量是上游任务（数据发送方）数量的整数倍时，rescale的效率明显会更高。比如当上游任务数量是2，下游任务数量是6时，上游任务其中一个分区的数据就将会平均分配到下游任务的3个分区中。
由于rebalance是所有分区数据的“重新平衡”，当 askManager数据量较多时，这种跨节点的网络传输必然影响效率；而如果我们配置的task slot数量合适，用rescale的方式进行“局 部重缩放”，就可以让数据只在当前TaskManager的多个slot之间重新分配，从而避免了网络传输带来的损耗。
从底层实现上看，rebalance和rescale的根本区别在于任务之间的连接机制不同。rebalance将会针对所有上游任务（发送数据方）和所有下游任务（接收数据方）之间建立通信通道，这是一个笛卡尔积的关系；而rescale仅仅针对每一个任务和下游对应的部分任务之间建立通信通道，节省了很多资源。可以在代码中测试如下：
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
30public class RescaleTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 这里使用了并行数据源的富函数版本
// 这样可以调用 getRuntimeContext 方法来获取运行时上下文的一些信息
env.addSource(new RichParallelSourceFunction<Integer>() {
public void run(SourceContext<Integer> sourceContext) throws
Exception {
for (int i = 1; i <= 8; i++) {
// 将奇数发送到索引为 1 的并行子任务
// 将偶数发送到索引为 0 的并行子任务
if (i % 2 == getRuntimeContext().getIndexOfThisSubtask()) {
sourceContext.collect(i);
}
}
}
public void cancel() {
}
})
.setParallelism(2)
.rescale()
.print().setParallelism(4);
env.execute();
}
}![]()
4.1.4 广播(broadcast)
这种方式其实不应该叫做“重分区”，因为经过广播之后，数据会在不同的分区都保留一份，可能进行重复处理。可以通过调用DataStream的broadcast()方法，将输入数据复制并发送 到下游算子的所有并行任务中去。 具体代码测试如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16public class BroadcastTest {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 读取数据源，并行度为 1
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 1200L)
);
// 经广播后打印输出，并行度为 4
stream.broadcast().print("broadcast").setParallelism(4);
env.execute();
}
}![]()
可以看到，数据被复制然后广播到了下游的所有并行任务中去了。
4.1.5 全局分区(global)
- 全局分区也是一种特殊的分区方式。这种做法非常极端，通过调用.global()方法，会将所有的输入流数据都发送到下游算子的第一个并行子任务中去。这就相当于强行让下游任务并行 度变成了1，所以使用这个操作需要非常谨慎，可能对程序造成很大的压力。
4.1.6 自定义分区(Custom)
当Flink提供的所有分区策略都不能满足用户的需求时，我们可以通过使用partitionCustom()方法来自定义分区策略。
在调用时，方法需要传入两个参数，第一个是自定义分区器（Partitioner）对象，第二个是应用分区器的字段，它的指定方式与keyBy指定key基本一样：可以通过字段名称指定，也可以通过字段位置索引来指定，还可以实现一个KeySelector。例如，我们可以对一组自然数按照奇偶性进行重分区。代码如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22public class CustomPartitionTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 将自然数按照奇偶分区
env.fromElements(1, 2, 3, 4, 5, 6, 7, 8)
.partitionCustom(new Partitioner<Integer>() {
public int partition(Integer key, int numPartitions) {
return key % 2;
}
}, new KeySelector<Integer, Integer>() {
public Integer getKey(Integer value) throws Exception {
return value;
}
})
.print().setParallelism(2);
env.execute();
}
}![]()
5、输出算子(Sink)
Flink作为数据处理框架，最终还是要把计算处理的结果写入外部存储，为外部应用提供支持，如图所示，本节将主要讲解Flink中的Sink操作。我们已经了解了Flink程序如何对数据进行读取、转换等操作，最后一步当然就应该将结果数据保存或输出到外部系统了。
![]()
5.1 连接到外部系统
Flink的DataStream API专门提供了向外部写入数据的方法：addSink。与addSource类似，addSink方法对应着一个“Sink”算子，主要就是用来实现与外部系统连接、并将数据提交写入的；Flink程序中所有对外的输出操作，一般都是利用Sink算子完成的。
Sink一词有“下沉”的意思，有些资料会相对于“数据源”把它翻译为“数据汇”。不论怎样理解，Sink在Flink中代表了将结果数据收集起来、输出到外部的意思，所以我们这里统一把它直观地叫作“输出算子”。
之前我们一直在使用的print方法其实就是一种Sink，它表示将数据流写入标准控制台打印输出。查看源码可以发现，print方法返回的就是一个DataStreamSink。
1
2
3
4
5public DataStreamSink<T> print(String sinkIdentifier) {
PrintSinkFunction<T> printFunction = new PrintSinkFunction<>(sinkIdentifier,
false);
return addSink(printFunction).name("Print to Std. Out");
}与Source算子非常类似，除去一些Flink预实现的Sink，一般情况下Sink算子的创建是通过调用DataStream的.addSink()方法实现的。
1
stream.addSink(new SinkFunction(...));
addSource的参数需要实现一个SourceFunction接口;类似地，addSink方法同样需要传入一个参数，实现的是 SinkFunction 接口。在这个接口中只需要重写一个方法invoke(),用来将指定的值写入到外部系统中。这个方法在每条数据记录到来时都会调用:
1
default void invoke(IN value, Context context) throws Exception
当然，SinkFuntion多数情况下同样并不需要我们自己实现。Flink官方提供了一部分的框架的Sink连接器。如图所示，列出了Flink官方目前支持的第三方系统连接器:
![image-20220905140858059]()
我们可以看到，像Kafka之类流式系统，Flink提供了完美对接，source/sink两端都能连接，可读可写;而对于Elasticsearch、文件系统(FileSystem)、JDBC等数据存储系统，则只提供了输出写入的sink连接器。除Flink官方之外，Apache Bahir作为给Spark和Flink提供扩展支持的项目，也实现了一 些其他第三方系统与Flink的连接器，如图所示。
![image-20220905141002650]()
除此以外，就需要用户自定义实现sink连接器了。
5.2 输出到文件
最简单的输出方式，当然就是写入文件了。对应着读取文件作为输入数据源，Flink本来也有一些非常简单粗暴的输出到文件的预实现方法:如writeAsText()、writeAsCsv()，可以直接将输出结果保存到文本文件或Csv文件。但我们知道，这种方式是不支持同时写入一份文件的;所以我们往往会将最后的Sink操作并行度设为1，这就大大拖慢了系统效率;而且对于故障恢复后的状态一致性，也没有任何保证。所以目前这些简单的方法已经要被弃用。
Flink为此专门提供了一个流式文件系统的连接器:StreamingFileSink，它继承自抽象类RichSinkFunction，而且集成了Flink的检查点(checkpoint)机制，用来保证精确一次(exactly once)的一致性语义。它的主要操作是将数据写入桶(buckets)，每个桶中的数据都可以分割成一个个大小有限的分区文件，这样一来就实现真正意义上的分布式文件存储。我们可以通过各种配置来控制“分桶” 的操作;默认的分桶方式是基于时间的，我们每小时写入一个新的桶。换句话说，每个桶内保存的文件，记录的都是1小时的输出数据。
StreamingFileSink支持行编码(Row-encoded)和批量编码(Bulk-encoded，比如Parquet) 格式。这两种不同的方式都有各自的构建器(builder)，调用方法也非常简单，可以直接调用StreamingFileSink的静态方法:
- 行编码:StreamingFileSink.forRowFormat(basePath，rowEncoder)。
- 批量编码:StreamingFileSink.forBulkFormat(basePath，bulkWriterFactory)。
在创建行或批量编码Sink时，我们需要传入两个参数，用来指定存储桶的基本路径(basePath)和数据的编码逻辑(rowEncoder或bulkWriterFactory)。下面我们就以行编码为例，将一些测试数据直接写入文件:
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
31public class SinkToFileTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Alice", "./prod?id=100", 3000L),
new Event("Alice", "./prod?id=200", 3500L),
new Event("Bob", "./prod?id=2", 2500L),
new Event("Alice", "./prod?id=300", 3600L),
new Event("Bob", "./home", 3000L),
new Event("Bob", "./prod?id=1", 2300L),
new Event("Bob", "./prod?id=3", 3300L));
StreamingFileSink<String> fileSink = StreamingFileSink
.<String>forRowFormat(new Path("./output"),
new SimpleStringEncoder<>("UTF-8"))
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
.withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
.withMaxPartSize(1024 * 1024 * 1024)
.build()).build();
// 将 Event 转换成 String 写入文件
stream.map(Event::toString).addSink(fileSink);
env.execute();
}
}- 这里我们创建了一个简单的文件Sink，通过.withRollingPolicy()方法指定了一个“滚动策略”。“滚动”的概念在日志文件的写入中经常遇到:因为文件会有内容持续不断地写入，所以我们应该给一个标准，到什么时候就开启新的文件，将之前的内容归档保存。也就是说，上面的代码设置了在以下3种情况下，我们就会滚动分区文件:
- 至少包含15分钟的数据。
- 最近5分钟没有收到新的数据。
- 文件大小已达到1GB。
![]( "image-20220905144258446")
- 这里我们创建了一个简单的文件Sink，通过.withRollingPolicy()方法指定了一个“滚动策略”。“滚动”的概念在日志文件的写入中经常遇到:因为文件会有内容持续不断地写入，所以我们应该给一个标准，到什么时候就开启新的文件，将之前的内容归档保存。也就是说，上面的代码设置了在以下3种情况下，我们就会滚动分区文件:
5.3 输出到Kafka
Kafka是一个分布式的基于发布/订阅的消息系统，本身处理的也是流式数据，所以跟Flink“天生一对”，经常会作为Flink的输入数据源和输出系统。Flink官方为Kafka提供了Source和Sink的连接器，我们可以用它方便地从Kafka读写数据。如果仅仅是支持读写，那还说明不了Kafka和Flink关系的亲密;真正让它们密不可分的是，Flink与Kafka的连接器提供了端到端的精确一次(exactly once)语义保证，这在实际项目中是最高级别的一致性保证。现在我们要将数据输出到Kafka，整个数据处理的闭环已经形成，所以可以完整测试如下:
- 添加Kafka连接器依赖。在之前测试过从Kafka数据源读取数据，连接器相关依赖已经引入。
- 启动Kafka集群。
- 编写输出到Kafka的示例代码。
我们可以直接将用户行为数据保存为文件clicks.csv,读取后不做转换直接写入Kafka，主题(topic)命名为“clicks”。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16public class SinkToKafkaTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
Properties properties = new Properties();
properties.put("bootstrap.servers", "localhost:9092");
DataStreamSource<String> stream = env.readTextFile("input/clicks.csv");
stream
.addSink(new FlinkKafkaProducer<String>(
"clicks",
new SimpleStringSchema(),
properties
));
env.execute();
}
}这里我们可以看到，addSink传入的参数是一个FlinkKafkaProducer。这也很好理解，因为需要向Kafka写入数据，自然应该创建一个生产者。FlinkKafkaProducer继承了抽象类TwoPhaseCommitSinkFunction，这是一个实现了“两阶段提交”的RichSinkFunction。两阶段提交提供了Flink向Kafka写入数据的事务性保证，能够真正做到精确一次(exactly once)的状态一致性。
运行代码，在Linux主机启动一个消费者, 查看是否收到数据。
1
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic clicks
5.4 输出到Redis
Flink没有直接提供官方的Redis连接器，不过Bahir项目还是担任了合格的辅助角色，为我们提供了Flink-Redis的连接工具。但版本升级略显滞后，目前连接器版本为1.0，支持的Scala版本最新到2.11。由于我们的测试不涉及到Scala的相关版本变化，所以并不影响使用。在实际项目应用中，应该以匹配的组件版本运行。
导入的Redis连接器依赖。
1
2
3
4
5<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_2.11</artifactId>
<version>1.0</version>
</dependency>启动Redis服务。
编写输出到Redis的示例代码。
连接器为我们提供了一个RedisSink，它继承了抽象类RichSinkFunction，这就是已经实现好的向Redis写入数据的SinkFunction。我们可以直接将Event数据输出到Redis:
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
27public class SinkToRedisTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 创建一个到 redis 连接的配置
FlinkJedisPoolConfig conf= new FlinkJedisPoolConfig.Builder().setHost("hadoop102").build();
env.addSource(new ClickSource())
.addSink(new RedisSink<Event>(conf, new MyRedisMapper()));
env.execute();
}
public static class MyRedisMapper implements RedisMapper<Event> {
public String getKeyFromData(Event e) {
return e.user;
}
public String getValueFromData(Event e) {
return e.url;
}
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.HSET, "clicks");
}
}
}这里RedisSink的构造方法需要传入两个参数:
- JFlinkJedisConfigBase：Jedis的连接配置。
- RedisMapper：Redis映射类接口，说明怎样将数据转换成可以写入Redis的类型。
在这里我们可以看到，保存到Redis时调用的命令是HSET，所以是保存为哈希表(hash)，表名为“clicks”；保存的数据以user 为key，以url为value，每来一条数据就会做一次转换。
运行代码，Redis查看是否收到数据。
1
2
3
4
5
6$ redis-cli
localhost:6379>hgetall clicks
1) “Mary”
2) “./home”
3) “Bob”
4) “./cart”我们会发现，发送了多条数据，Redis中只有2条数据，原因是hash中的key重复了，后面的会把前面的覆盖掉。
5.5 输出到Elasticsearch
ElasticSearch是一个分布式的开源搜索和分析引擎，适用于所有类型的数据。ElasticSearch有着简洁的REST风格的API，以良好的分布式特性、速度和可扩展性而闻名，在大数据领域应用非常广泛。
Flink为ElasticSearch专门提供了官方的Sink连接器，Flink 1.13支持当前最新版本的ElasticSearch。写入数据的ElasticSearch的测试步骤如下。
添加Elasticsearch连接器依赖。
1
2
3
4
5<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-elasticsearch7_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>启动Elasticsearch服务。
编写输出到Elasticsearch的示例代码。
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
35public class SinkToEsTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Alice", "./prod?id=100", 3000L),
new Event("Alice", "./prod?id=200", 3500L),
new Event("Bob", "./prod?id=2", 2500L),
new Event("Alice", "./prod?id=300", 3600L),
new Event("Bob", "./home", 3000L),
new Event("Bob", "./prod?id=1", 2300L),
new Event("Bob", "./prod?id=3", 3300L));
ArrayList<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("localhost", 9200, "http"));
// 创建一个 ElasticsearchSinkFunction
ElasticsearchSinkFunction<Event> elasticsearchSinkFunction = new ElasticsearchSinkFunction<Event>() {
public void process(Event element, RuntimeContext ctx, RequestIndexer indexer) {
HashMap<String, String> data = new HashMap<>();
data.put(element.user, element.url);
IndexRequest request = Requests.indexRequest()
.index("clicks")
.type("type") // Es 6 必须定义 type
.source(data);
indexer.add(request);
}
};
stream.addSink(new ElasticsearchSink.Builder<Event>(httpHosts, elasticsearchSinkFunction).build());
env.execute();
}
}- 与RedisSink类似，连接器也为我们实现了写入到Elasticsearch的SinkFunction——ElasticsearchSink。区别在于，这个类的构造方法是私有(private)的，我们需要使用ElasticsearchSink的Builder内部静态类，调用它的build()方法才能创建出真正的 SinkFunction。而Builder的构造方法中又有两个参数:
- httpHosts：连接到的Elasticsearch集群主机列表。
- elasticsearchSinkFunction：这并不是我们所说的SinkFunction，而是用来说明具体处理逻辑、准备数据向 Elasticsearch发送请求的函数。
- 具体的操作需要重写中elasticsearchSinkFunction中的process方法，我们可以将要发送的数据放在一个HashMap中，包装成IndexRequest向外部发送HTTP请求。
- 与RedisSink类似，连接器也为我们实现了写入到Elasticsearch的SinkFunction——ElasticsearchSink。区别在于，这个类的构造方法是私有(private)的，我们需要使用ElasticsearchSink的Builder内部静态类，调用它的build()方法才能创建出真正的 SinkFunction。而Builder的构造方法中又有两个参数:
运行代码，访问Elasticsearch查看是否收到数据，查询结果如下所示。
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{
"took" : 5,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
}, "hits" :
"total" : {
"value" : 9,
"relation" : "eq"
},
"max_score" : 1.0,
"hits" : [
{
"_index" : "clicks",
"_type" : "_doc",
"_id" : "dAxBYHoB7eAyu-y5suyU",
"_score" : 1.0,
"_source" : {
"Mary" : "./home"
}
}
... ]
} }
5.6 输出到MySQL(JDBC)
关系型数据库有着非常好的结构化数据设计、方便的SQL查询，是很多企业中业务数据存储的主要形式。MySQL就是其中的典型代表。尽管在大数据处理中直接与MySQL交互的场景不多，但最终处理的计算结果是要给外部应用消费使用的，而外部应用读取的数据存储往往就是MySQL。所以我们也需要知道如何将数据输出到MySQL这样的传统数据库。写入数据的MySQL的测试步骤如下。
添加依赖。
1
2
3
4
5
6
7
8
9
10<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.47</version>
</dependency>启动MySQL，在database库下建表clicks。
1
2
3mysql> create table clicks(
-> user varchar(20) not null,
-> url varchar(100) not null);编写输出到MySQL的示例代码。
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
40public class SinkToMySQL {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStreamSource<Event> stream = env.fromElements(
new Event("Mary", "./home", 1000L),
new Event("Bob", "./cart", 2000L),
new Event("Alice", "./prod?id=100", 3000L),
new Event("Alice", "./prod?id=200", 3500L),
new Event("Bob", "./prod?id=2", 2500L),
new Event("Alice", "./prod?id=300", 3600L),
new Event("Bob", "./home", 3000L),
new Event("Bob", "./prod?id=1", 2300L),
new Event("Bob", "./prod?id=3", 3300L));
stream.addSink(
JdbcSink.sink(
"INSERT INTO clicks (user, url) VALUES (?, ?)",
(statement, r) -> {
statement.setString(1, r.user);
statement.setString(2, r.url);
},
JdbcExecutionOptions.builder()
.withBatchSize(1000)
.withBatchIntervalMs(200)
.withMaxRetries(5)
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://localhost:3306/userbehavior")
// 对于 MySQL 5.7，用"com.mysql.jdbc.Driver"
.withDriverName("com.mysql.cj.jdbc.Driver")
.withUsername("username")
.withPassword("password")
.build()
)
);
env.execute();
}
}运行代码，用客户端连接MySQL，查看是否成功写入数据。
1
2
3
4
5
6
7
8
9mysql> select * from clicks;
+------+--------------+
| user | url |
+------+--------------+
| Mary | ./home |
| Alice| ./prod?id=300|
| Bob | ./prod?id=3 |
+------+--------------+
3 rows in set (0.00 sec)
5.7 自定义Sink输出
与Source类似，Flink为我们提供了通用的SinkFunction接口和对应的RichSinkDunction抽象类，只要实现它，通过简单地调用 DataStream的.addSink()方法就可以自定义写入任何外部存储。之前与外部系统的连接，其实都是连接器帮我们实现了SinkFunction，现在既然没有现成的，我们就只好自力更生了。例如，Flink并没有提供HBase的连接器，所以需要我们自己写。
在实现SinkFunction的时候，需要重写的一个关键方法invoke()，在这个方法中我们就可以实现将流里的数据发送出去的逻辑。我们这里使用了SinkFunction的富函数版本，因为这里我们又使用到了生命周期的概念， 创建HBase的连接以及关闭HBase的连接需要分别放在open()方法和close()方法中。
导入依赖。
1
2
3
4
5<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>3.0.0-alpha-3</version>
</dependency>编写输出到HBase的示例代码。
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
42
43
44public class SinkCustomtoHBase {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env
.fromElements("hello", "world")
.addSink(
new RichSinkFunction<String>() {
public org.apache.hadoop.conf.Configuration
configuration; // 管理 Hbase 的配置信息,这里因为 Configuration 的重名问题，将类以完整路径 导入
public Connection connection; // 管理 Hbase 连接
public void open(Configuration parameters) throws Exception {
super.open(parameters);
configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum",
"localhost:2181");
connection = ConnectionFactory.createConnection(configuration);
}
public void invoke(String value, Context context) throws Exception {
Table table = connection.getTable(TableName.valueOf("test")); // 表名为 test
Put put = new Put("rowkey" .getBytes(StandardCharsets.UTF_8)); // 指定 rowkey
put.addColumn("info" .getBytes(StandardCharsets.UTF_8) // 指定列名
, value.getBytes(StandardCharsets.UTF_8) // 写入的数据
, "1" .getBytes(StandardCharsets.UTF_8)); // 写入的数据
table.put(put); // 执行 put 操作
table.close(); // 将表关闭
}
public void close() throws Exception {
super.close();
connection.close(); // 关闭连接
}
}
);
env.execute();
}
}可以在HBase查看插入的数据。