Flink遇到KafkaFlinkKafkaConsumer使用详解

Flink是新一代的流处理计算引擎。通过轻量级的checkpoint&＃xff0c;Flink可以在高吞吐量的情况下保证exactly-once(这需要数据源能够提供回溯消费的能力)。Flink支持众多的source(从中读取数据)和sink&＃xff08;向其写入数据&＃xff09;&＃xff0c;列表如下&＃xff1a;

Kafka作为目前非常流行的消息中间件&＃xff0c;它不仅能够提供极大的吞吐量&＃xff0c;还能够配合Flink在消费端达到exactly-once。

本文将详细介绍如何配置Flink读取Kafka&＃xff0c;运行机制和exactly-once是如何保证的&＃xff0c;最后&＃xff0c;还会给出监控Flink消费Kafka的方案。(注: 本文的使用的是Flink 1.3.1-release和 Kafka 0.8)

Flink 是通过Connector与具体的source 和 sink进行通信的&＃xff0c;具体到Kafka 0.8&＃xff0c;相应的Connector是 FlinkKafkaConsumer08和FlinkKafkaProducer08。

我们首先介绍FlinkKafkaConsumer08的配置&＃xff1a;

一、 Kafka Consumer的配置

FlinkKafkaConsumer08可以消费一个或多个Kafka topic的数据&＃xff0c;它的构造器需要接收以下参数:

1. topic名或 topic名的列表

2. 反序列化约束&＃xff0c;以便于Flink决定如何反序列化从Kafka获得的数据

3. Kafka consumer的属性配置&＃xff0c;下面两个属性配置是必须的:

· “zookeeper.connect” (Zookeeper servers的地址列表&＃xff0c;以逗号分隔)

· “group.id” (consumer group)

· “bootstrap.servers” (Kafka brokers的地址列表&＃xff0c;以逗号分隔)

示例代码&＃xff1a;

以下几个参数是需要我们重点关注的。

(一) 反序列化shema

Flink Kafka Consumer 需要知道如何将来自Kafka的二进制数据转换为Java/Scala对象。DeserializationSchema接口允许程序员指定这个序列化的实现。该接口的 T deserialize(byte[]message) 会在收到每一条Kafka的消息的时候被调用。

我们通常会实现 AbstractDeserializationSchema&＃xff0c;它可以描述被序列化的Java/Scala类型到Flink的类型(TypeInformation)的映射。如果用户的代码实现了DeserializationSchema&＃xff0c;那么就需要自己实现getProducedType(…) 方法。

为了方便使用&＃xff0c;Flink提供了一些已实现的schema:

1. TypeInformationSerializationSchema (andTypeInformationKeyValueSerializationSchema) &＃xff0c;他们会基于Flink的TypeInformation来创建schema。这对于那些从Flink写入&＃xff0c;又从Flink读出的数据是很有用的。这种Flink-specific的反序列化会比其他通用的序列化方式带来更高的性能。

2. JsonDeserializationSchema (andJSONKeyValueDeserializationSchema) 可以把序列化后的Json反序列化成ObjectNode&＃xff0c;ObjectNode可以通过objectNode.get(“field”).as(Int/String/…)() 来访问指定的字段。

3. SimpleStringSchema可以将消息反序列化为字符串。当我们接收到消息并且反序列化失败的时候&＃xff0c;会出现以下两种情况: 1) Flink从deserialize(…)方法中抛出异常&＃xff0c;这会导致job的失败&＃xff0c;然后job会重启&＃xff1b;2) 在deserialize(…) 方法出现失败的时候返回null&＃xff0c;这会让Flink Kafka consumer默默的忽略这条消息。请注意&＃xff0c;如果配置了checkpoint 为enable&＃xff0c;由于consumer的失败容忍机制&＃xff0c;失败的消息会被继续消费&＃xff0c;因此还会继续失败&＃xff0c;这就会导致job被不断自动重启。

(二) Kafka Consumers 起始offset配置

FlinkKafkaConsumer 允许我们配置Kafka partition被消费的offset的起始位&＃xff0c;示例代码如下&＃xff1a;

所有版本的Flink KafkaConsumer都支持以上的配置&＃xff0c;下面对这些配置进行详细的说明&＃xff1a;

setStartFromGroupOffsets(默认)&＃xff1a;采用consumer group的offset来作为起始位&＃xff0c;这个offset从Kafka brokers(0.9以上版本) 或 Zookeeper(Kafka 0.8)中获取。如果从Kafka brokers或者Zookeeper中找不到这个consumer group对应的partition的offset&＃xff0c;那么auto.offset.reset这个配置就会被启用。

setStartFromEarliest() /setStartFromLatest(): 即从最早的/最新的消息开始消费。

当然&＃xff0c;也可以指定具体的某个offset作为某个partition的起始消费位置:

上述的代码配置了myTopic的partition 0,1,2在被Flink job消费的起始位置。假设myTopic总共有5个partition&＃xff0c;那么剩下的两个partition没有被配置具体的offset的起始位&＃xff0c;所以Flink会对这两个partition的采用默认的offset起始位的配置(setStartFromGroupOffsets)。

注意&＃xff0c;如果你在这个job中配置了enableCheckpointing() 或者从某个savepoint来启动这个job&＃xff0c;那么起始位会优先从savepoint或者checkpoint中获取。

(三) 容错机制

当Flink的job开启了checkpoint的时候&＃xff0c;Flink会一边消费topic的数据&＃xff0c;一边定时的将offset和其他operator的状态记录到checkpoint中。如果遇到了job失败的情况&＃xff0c;那么Flink将会重启job&＃xff0c;从最后一个checkpoint中来恢复job的所有状态&＃xff0c;然后从checkpoint中记录的offset开始重新对Kafka 的topic进行消费。记录offset的间隔决定了程序在失败的情况下需要回溯的最大程度。

为了使用Flink Kafkaconsumer的容错机制&＃xff0c;我们需要在程序中作如下的配置&＃xff1a;

还有一点需要注意的是&＃xff0c;Flink只有在task slot的数量足够的情况下才可以成功的重启job&＃xff0c;所以如果job是因为TaskManager down掉&＃xff08;或者无法连接到集群&＃xff09;导致task slot不足而失败&＃xff0c;那么必须要恢复增加足够的task slot才能让job重启。而Flink on YARN 支持自动的重启丢失的YARN containers。

(四) offset提交行为的配置

Flink KafkaConsumer允许配置向 Kafka brokers&＃xff08;或者向Zookeeper&＃xff09;提交offset的行为。需要注意的是&＃xff0c;Flink Kafka Consumer并不依赖于这些提交回Kafka或Zookeeper的offset来保证容错。这些被提交的offset只是意味着Flink将消费的状态暴露在外以便于监控。

Checkpointingdisabled: 此时&＃xff0c; Flink Kafka Consumer依赖于它使用的具体的Kafka client的自动定期提交offset的行为&＃xff0c;相应的设置是 Kafka properties中的 enable.auto.commit (或者 auto.commit.enable 对于Kafka 0.8) 以及 auto.commit.interval.ms。

Checkpointingenabled: 在这种情况下&＃xff0c;Flink Kafka Consumer会将offset存到checkpoint中当checkpoint 处于completed的状态时。这保证了在Kafka brokers中的committed offset和checkpointed states中的offset保持一致。通过调用setCommitOffsetOnCheckpoints(boolean)来调整 offset自动提交是否开启(默认情况下是true&＃xff0c;即开启自动提交)。请注意&＃xff0c;在这种情况下&＃xff0c;配置在properties 中的offset的定时自动提交行为将会被忽略。

二、Flink Kafka Consumer的运行机制

上图简要概括了FlinkKafkaConsumer08的运行机制。每个消费Kafka source的operator的subTask线程都持有一个FlinkKafkaConsumer08实例&＃xff0c;这个实例负责分配这个subTask线程消费的topic的具体的partition&＃xff0c;以及从checkpoint中恢复partition应该消费的起始offset。

Kafka08Fetcher负责和Kafka brokers通信&＃xff0c;获取具体各个partition的leader&＃xff0c;每个FlinkKafkaConsumer08都拥有一个Kafka08Fetcher。每个Kafka08Fetcher拥有一个或多个SimpleConsumerThread&＃xff0c;SimpleConsumerThread负责从partition的leader中拉取数据&＃xff0c;并将其反序列化&＃xff0c;最后发送给下一级的operator&＃xff0c;注意在SimpleConsumerThread中使用的是Kafka的低级API&＃xff0c;这是因为它需要灵活的控制从某个具体的offset进行消费。

(一) 生命周期

下面&＃xff0c;我们罗列出FlinkKafkaConsumer08、Kafka08Fetcher和SimpleConsumerThread生命周期的几个关键点&＃xff08;关于每个Flink StreamTask的生命周期&＃xff0c;可以参考&＃xff1a;https://ci.apache.org/projects/flink/flink-docs-release-1.3/internals/task_lifecycle.html&＃xff09;&＃xff0c;以弄清楚FlinkKafkaConsumer08是如何恢复offset 以及保证exactly-once的。

1.FlinkKafkaConsumer08&＃xff1a;

1. initializeState()&＃xff1a;从最后一个成功的checkpoint中获取各个partition的offset到restoredState中。

2. open()&＃xff1a;从restoredState中获取这个subTask所消费的topic的partition的起始offset&＃xff0c;保存到subscribedPartitionsToStartOffsets中&＃xff1b;如果这是一个第一次向topic消费的job的subTask&＃xff0c;那么Flink根据job的并行度以及这个subTask的index均匀的分配partition给这个subTask消费。此时&＃xff0c;partition的起始offset就由我们在上文中介绍的配置来决定。

3. run(): 如果subscribedPartitionsToStartOffsets不为空&＃xff0c;创建Kafka08Fetcher&＃xff0c;执行其runFetchLoop()。

4. close(): job被cancel或者出现了异常&＃xff0c;那么调用close()方法&＃xff0c;close方法会调用Fetcher的cancel()方法&＃xff0c;FlinkKafkaConsumer08所在的subTask结束。

2. Kafka08Fetcher:

Kafka08Fetcher从FlinkKafkaConsumer08中获取它要消费的partition加入到它的unassignedPartitionsQueue中。然后在它的runFetchLoop()中&＃xff0c;对这个Queue中的partition进行消费。下面主要罗列runFetchLoop()中的主要细节。

1. 首先创建brokerToThread:Map&＃xff0c;这个map会把broker和连接到这个broker的线程映射起来。创建zookeeperOffsetHandler&＃xff0c;根据配置的起始的offset行为从zookeeper或Kafka中获取没有从checkpoint中恢复的partition的offset。然后创建PeriodicOffsetCommitter线程周期性的向Zookeeper提交offset。

2. 向Flink内部(JMX)注册MetricGroup&＃xff0c;名为"KafkaConsumer"&＃xff0c;我们可以利用这个MetricGroup来对FlinkKafakConsumer08的消费状况进行监控。

3. while(running) 循环&＃xff1a;获取unassignedPartitionsQueue中应该被消费的partition&＃xff0c;通过低级api寻找他们的Leader。然后从brokerToThread中寻找这个leader对应的SimpleConsumerThread是否被创建了&＃xff0c;如果没有被创建&＃xff0c;那么创建并运行一个SimpleConsumerThread&＃xff0c;并更新brokerToThread&＃xff1b;如果已经被创建&＃xff0c;并且没有新的partition加入&＃xff0c;那么重复这个while循环。否则&＃xff0c;终止这个leader对应的SimpleConsumerThread&＃xff0c;并创建新的SimpleConsumerThread然后继续消费。

4. cancel():如果发生了异常或者job被手动cancel&＃xff0c;关闭zookeeperOffsetHandler、periodicCommitter和brokerToThread中的所有线程。

3.SimpleConsumerThread:

1. 创建SimpleConsumer(Kafka低级api)&＃xff0c;然后再一次根据配置的起始offset行为去Kafka中获取没有从checkpoint和Zookeeper中恢复的partition的offset。

2. while(running) 循环&＃xff0c;我们跳过errorhandle&＃xff0c;直接展示最核心的代码:
   

其中valueBytes和keyBytes均为从kafka中读取到的消息&＃xff0c;他们经过反序列化后交给owner发送消息给下一级的operator。

这里的owner就是Kafka08Fetcher&＃xff0c;它除了会将消息发送给下一级的operator外&＃xff0c;还会记录这个partition的state(即offset)&＃xff0c;Flink异步的将这个state记录到checkpoint中。

这里需要注意的是&＃xff0c;Flink异步记录checkpoint的行为是由我们的来配置的&＃xff0c;只有当我们设置了enableCheckpointing()时&＃xff0c;Flink才会在checkpoint完成时(整个job的所有的operator都收到了这个checkpoint的barrier才意味这checkpoint完成&＃xff0c;具体参考我们对Flink checkpoint的介绍)将offset记录起来并提交&＃xff0c;这时候才能够保证exactly-once。

3. 如果线程被终止&＃xff0c;那么关闭SimpleConsumer。

(二) 容错机制

在发生错误的情况下&＃xff0c;Flink会如何处理呢&＃xff1f;在finally块中记录最后消费到的offset再向JobManager提交checkpoint吗&＃xff1f;在通常情况下&＃xff0c;比如发生了手动cancel或者userCode的异常时&＃xff0c;这么做没有问题。可是如果是因为其他原因(如Full GC)使得TaskManagerhung住了&＃xff0c;甚至是机器挂了&＃xff0c;那么这个时候就不能通过finally 块来保证exactly-once了。Flink依赖的是带barrier的checkpointing机制来解决容错的问题。

我们通过下面一副图来简述这种机制:

barrier可以理解为checkpoint之间的分隔符&＃xff0c;在它之前的data属于前一个checkpoint&＃xff0c;而在它之后的data属于另一个checkpoint。同时&＃xff0c;barrier会由source(如FlinkKafkaConsumer)发起&＃xff0c;并混在数据中&＃xff0c;同数据一样传输给下一级的operator&＃xff0c;直到sink为止。假设我们的Streaming Job只有一个source、一个map operator 以及一个sink&＃xff0c;属于barrier所分隔的checkpoint 的数据已经被处理完毕并sink&＃xff0c;而barrier还处于source和map operator之间&＃xff0c;barrier 还处于map和sink之间。由于barrier已经被sink收到&＃xff0c;那么说明checkpoint已经完成了(这个checkpoint的状态为completed并被存到了state backend中)&＃xff0c;它之前的数据已经被处理完毕并sink。

但是由于sink还没有收到barrier&＃xff0c;那么所有之前之后的数据都会被缓存在sink的Input Buffer中&＃xff0c;也就是说这部分数据虽然已经经过source消费并经过map处理了&＃xff0c;但是还是没有写入目的地。所以如果Job在这个时候失败了&＃xff0c;最后一个成功committed的checkpoint是checkpoint&＃xff0c;所以FlinkKafkaConsumer从checkpoint中恢复出相应的partitionoffset就可以了。

我们注意到&＃xff0c;虽然之后的部分数据和之后的所有数据虽然已经被source消费&＃xff0c;但是都没有被sink&＃xff0c;这部分数据会被FlinkKafkaConsumer“重复”消费&＃xff0c;我们并没有丢失任何的数据也没有重复写入任何数据&＃xff0c;保证了exactly-once。

小节&＃xff1a;

1. 在配置了checkpointingenable的情况下&＃xff0c;FlinkKafkaConsumer08在开始消费数据之前&＃xff0c;会优先从checkpoint中恢复出被消费的partition的offset&＃xff0c;如果没有从checkpoint中恢复某些partition的offset&＃xff0c;它会从Zookeeper中恢复&＃xff0c;若从Zookeeper中仍然没有恢复&＃xff0c;它会根据配置的offset起始行为来配置起始offset。

2.FlinkKafkaConsumer08通过Kafka的低级API和Flink带barrier的轻量级checkpoint机制保证了在高吞吐量的情况下的exactly-once。

三、监控

我们在前文提到&＃xff0c;Flink会定时的将offset提交到Zookeeper中&＃xff0c;但是提交到Zookeeper的offset并不是实时的offset。官方更为推荐从Flink注册的Metric来监控Flink的消费情况。

上面的两幅图片显示了analytics_package_standard这个kafka topic的 partition-0在Flink中的当前offset&＃xff0c;这个offset我们可以通过jconsole、jvisualvm等工具查看&＃xff0c;也可以直接通过Flink TaskManager开启的JMX 端口获取。如果我们要监控消费淤积&＃xff0c;再从Kafka中获取相应partition的latestoffset即可。我们可以在配置文件flink-conf.yaml中配置Flink的JMX端口&＃xff1a;

Flink 1.3中&＃xff0c;还增加了许多有用的监控&＃xff0c;比如总的消息条数&＃xff0c;消息的瞬时读取速度&＃xff0c;latency等等&＃xff0c;我们可以在Flink的Web UI的Task Metric中查看这些监控指标&＃xff1a;

结论

FlinkKafkaConsumer提供了一套健壮的机制保证了在高吞吐量的情况下exactly-once的消费Kafka的数据&＃xff0c;它的API的使用与配置也比较简单&＃xff0c;同时也便于监控。

转发链接&＃xff1a;https://www.sohu.com/a/168546400_617676