Flink(4)——source 介绍与实践

December 20, 2018
作者:星爷
出处:http://www.readingnotes.site/posts/Flink(4)-Source-%E4%BB%8B%E7%BB%8D%E4%B8%8E%E5%AE%9E%E8%B7%B5.html
声明:转载请注明作者及出处。

本文转自个人微信公众号,原文链接。本博客评论系统需要梯子,大家关注下公众号方便交流。

本文基于Apache Flink 1.7。

Source 就是Flink 程序的数据输入,Flink 提供了多种数据输入方式,下面逐一介绍。

概念

Flink预定义Sources

Flink 预定义了多种Sources。

Connectors

connectors 用于给接入第三方数据提供接口,现在支持的connectors 包括:

另外,通过 Apache Bahir,可以支持ActiveMQ/Netty之类的Source。

Async I/O API

Flink 提供了外部数据存储的异步I/O API。流计算中经常需要与外部存储系统交互,比如取某个表的数据以便跟流中数据进行关联,一般来说,如果用同步I/O的方式,会造成系统中出现大的等待时间,影响吞吐和延迟。为了解决这个问题,异步I/O可以并发处理多个请求,提高吞吐,减少延迟,如下图所示。

Async I/O

Queryable State

如果Flink 应用需要将大量数据写到外部存储,这时候很容易产生I/O 瓶颈,如果需要写的数据是读少写多的数据,那么是否可以让外部应用自己来拉取数据呢?Queryable State 就是这个用途,提供了接口给外部应用,允许外部应用根据需要查询Flink state,现阶段Queryable State 还是Beta版,期待ing。

容错

Flink 提供了容错机制,以便Jobs从Failure恢复并继续执行,Flink 提供source的 exactly-once需要source的支持,如下图所示(注:图片来源于Flink 官网):

Source 容错

实战

预定义Sources

预定义的Source比较简单,在程序开发、调试阶段,可以采用基于Collection的Source,举例来说:

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
val stream = env.fromCollection(List(1,2,3,4,5))
stream.print
env.execute

Kafka Connector

Connectors 就以最常用的Kafka Connectors来说。Flink 提供了Flink Kafka Consumer 读取Kafka topics的数据,Flink Kafka Consumer 集成了Flink的checkpoint 机制以提供exactly-once 语义,不仅以来Kafka Consumer 的offset 追踪,同时将这些信息存到checkpoint。

Kafka Connector 在Flink 1.7.0后有大的改动,但还处于beta阶段,所以,下面还是以flink-connector-kafka-0.11_2.11 为例,而且,我们在生产环境也是用的这个版本。大家也可以用最新的flink-connector-kafka_2.11,这是一个通用版本,兼容0.10.0后边的版本。

首先,在项目中import Flink Kafka Connector。

<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-connector-kafka-0.11_${scala.binary.version}</artifactId>
	<version>${flink.version}</version>
</dependency>

代码如下:

/** 注释1
 * 启动checkpoint(可选)
 * env.enableCheckpointing(5000); //checkpoint every 5000 msecs
 */
val properties = new Properties();
properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
properties.setProperty("group.id", "groupXXX");
/** 注释2
 * 下面配置读取kafka topic partition 的起始偏移量(可选)
 * consumer.setStartFromEarliest();
 * consumer.setStartFromLatest();
 * consumer.setStartFromTimestamp(...);
 * consumer.setStartFromGroupOffsets();  // 默认
*/
val consumer = new FlinkKafkaConsumer011[]("topic_name", SimpleStringSchema, properties)
/** 注释3
 * Watermark (可选)
 * consumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter());
 */
val stream = env.addSource(consumer)

1. 构造函数 ,需要三个参数:

2. checkpoint,从注释1 可以看出,Flink Kafka Consumer 可以启动checkpoint机制,会周期性的给Kafka offsets 和 Flink的其它states 做 checkpoints,当Job 失败时,Flink 读取checkpoint里最新的state并从对应offset 开始消费数据恢复运算。

3. 起始偏移量,从注释2可以看出,Flink Kafka Consumer 允许设置读取 Kafka Partition的起始偏移量,而且,允许不同Partitions 分别进行设置。但要注意,设置起始偏移量不适用于两种情况:

4. Kafka Topic 和Partition 自发现,比如构建Kafka Consumer时,topic名可以是正则表达式,这时候,如果有符合该正则的新的topic 加入到Kafka 集群,可以被自动发现;另外,如果对Kafka Topic 进行RePartition,也可以自动发现,使用不多,可以自行查阅文档。

5. Kafka Consumer与Watermark ,从注释3可以看出,结合上篇文章,可以给数据设置方法以便给数据带上watermark。

总结

本文主要以 Flink Kafka Connector 为例讲了Flink 里的Sources,主要是考虑Kafka广泛使用在实时系统中,甚至可以说是标配,后边将开始讲解Sink以及Flink SQL。

看到这里,请扫描下方二维码关注我,Happy Friday !

wx