Flink Kafka Connector與Exactly Once剖析

6

Flink Kafa Connector是Flink内置的Kafka連接器,它包含了從Kafka Topic讀入數據的Flink Kafka Consumer以及向Kafka Topic寫出數據的Flink Kafka Producer,除此之外Flink Kafa Connector基于Flink Checkpoint機制提供了完善的容錯能力。本文從Flink Kafka Connector的基本使用到Kafka在Flink中端到端的容錯原理展開讨論。

一、Flink Kafka的使用

在Flink中使用Kafka Connector時需要依賴Kafka的版本,Flink針對不同的Kafka版本提供了對應的Connector實現。

1.版本依賴

既然Flink對不同版本的Kafka有不同實現,在使用時需要注意區分,根據使用環境引入正确的依賴關系。

1<dependency>
2  <groupId>org.apache.flink</groupId>
3  <artifactId>${flink_kafka_connector_version}</artifactId>
4  <version>${flink_version}</version>
5</dependency>

在上面的依賴配置中${flink_version}指使用Flink的版本,${flink_connector_kafka_version}指依賴的Kafka connector版本對應的artifactId。下表描述了截止目前為止Kafka服務版本與Flink Connector之間的對應關系。
Flink官網内容Apache Kafka Connector(https://ci.apache.org/project...)中也有詳細的說明。

clipboard.png

從Flink 1.7版本開始為Kafka 1.0.0及以上版本提供了全新的的Kafka Connector支持,如果使用的Kafka版本在1.0.0及以上可以忽略因Kafka版本差異帶來的依賴變化。

2.基本使用

明确了使用的Kafka版本後就可以編寫一個基于Flink Kafka讀/寫的應用程序「本文讨論内容全部基于Flink 1.7版本和Kafka 1.1.0版本」。根據上面描述的對應關系在工程中添加Kafka Connector依賴。

1<dependency>
2  <groupId>org.apache.flink</groupId>
3  <artifactId>flink-connector-kafka_2.11</artifactId>
4  <version>1.7.0</version>
5</dependency>

下面的代碼片段是從Kafka Topic「flink_kafka_poc_input」中消費數據,再寫入Kafka Topic「flink_kafka_poc_output」的簡單示例。示例中除了讀/寫Kafka Topic外,沒有做其他的邏輯處理。

1public static void main(String[] args) {
 2  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 3
 4  /** 初始化Consumer配置 */
 5  Properties consumerConfig = new Properties();
 6  consumerConfig.setProperty("bootstrap.servers", "127.0.0.1:9091");
 7  consumerConfig.setProperty("group.id", "flink_poc_k110_consumer");
 8
 9  /** 初始化Kafka Consumer */
10  FlinkKafkaConsumer<String> flinkKafkaConsumer = 
11    new FlinkKafkaConsumer<String>(
12      "flink_kafka_poc_input", 
13      new SimpleStringSchema(), 
14      consumerConfig
15    );
16  /** 将Kafka Consumer加入到流處理 */
17  DataStream<String> stream = env.addSource(flinkKafkaConsumer);
18
19  /** 初始化Producer配置 */
20  Properties producerConfig = new Properties();
21  producerConfig.setProperty("bootstrap.servers", "127.0.0.1:9091");
22
23  /** 初始化Kafka Producer */
24  FlinkKafkaProducer<String> myProducer = 
25    new FlinkKafkaProducer<String>(
26      "flink_kafka_poc_output", 
27      new MapSerialization(), 
28      producerConfig
29    );
30  /** 将Kafka Producer加入到流處理 */
31  stream.addSink(myProducer);
32
33  /** 執行 */
34  env.execute();
35}
36
37class MapSerialization implements SerializationSchema<String> {
38  public byte[] serialize(String element) {
39    return element.getBytes();
40  }
41}

Flink API使用起來确實非常簡單,調用addSource方法和addSink方法就可以将初始化好的FlinkKafkaConsumerFlinkKafkaProducer加入到流處理中。execute執行後,KafkaConsumer和KafkaProducer就可以開始正常工作了。

二、Flink Kafka的容錯

衆所周知,Flink支持Exactly-once semantics。什麼意思呢?翻譯過來就是「恰好一次語義」。流處理系統中,數據源源不斷的流入到系統、被處理、最後輸出結果。我們都不希望系統因人為或外部因素産生任何意想不到的結果。對于Exactly-once語義達到的目的是指即使系統被人為停止、因故障shutdown、無故關機等任何因素停止運行狀态時,對于系統中的每條數據不會被重複處理也不會少處理。

1.Flink Exactly-once
Flink宣稱支持Exactly-once其針對的是Flink應用内部的數據流處理。但Flink應用内部要想處理數據首先要有數據流入到Flink應用,其次Flink應用對數據處理完畢後也理應對數據做後續的輸出。在Flink中數據的流入稱為Source,數據的後續輸出稱為Sink,對于Source和Sink完全依靠外部系統支撐(比如Kafka)。

Flink自身是無法保證外部系統的Exactly-once語義。但這樣一來其實并不能稱為完整的Exactly-once,或者說Flink并不能保證端到端Exactly-once。而對于數據精準性要求極高的系統必須要保證端到端的Exactly-once,所謂端到端是指Flink應用從Source一端開始到Sink一端結束,數據必經的起始和結束兩個端點。

那麼如何實現端到端的Exactly-once呢?Flink應用所依賴的外部系統需要提供Exactly-once支撐,并結合Flink提供的Checkpoint機制和Two Phase Commit才能實現Flink端到端的Exactly-once。對于Source和Sink的容錯保障,Flink官方給出了具體說明:

Fault Tolerance Guarantees of Data Sources and Sinks(https://ci.apache.org/project...

2.Flink Checkpoint
在讨論基于Kafka端到端的Exactly-once之前先簡單了解一下Flink Checkpoint,詳細内容在《Flink Checkpoint原理》中有做讨論。Flink Checkpoint是Flink用來實現應用一緻性快照的核心機制,當Flink因故障或其他原因重啟後可以通過最後一次成功的Checkpoint将應用恢複到當時的狀态。如果在應用中啟用了Checkpoint,會由JobManager按指定時間間隔觸發Checkpoint,Flink應用内所有帶狀态的Operator會處理每一輪Checkpoint生命周期内的幾個狀态。

  • initializeState:由CheckpointedFunction接口定義。Task啟動時獲取應用中所有實現了CheckpointedFunction的Operator,并觸發執行initializeState方法。在方法的實現中一般都是從狀态後端将快照狀态恢複。
  • snapshotState:由CheckpointedFunction接口定義。JobManager會定期發起Checkpoint,Task接收到Checkpoint後獲取應用中所有實現了CheckpointedFunction的Operator并觸發執行對應的snapshotState方法。JobManager每發起一輪Checkpoint都會攜帶一個自增的checkpointId,這個checkpointId代表了快照的輪次。
1public interface CheckpointedFunction {
2  void snapshotState(FunctionSnapshotContext context) throws Exception;
3  void initializeState(FunctionInitializationContext context) throws Exception;
4}
  • notifyCheckpointComplete:由CheckpointListener接口定義。當基于同一個輪次(checkpointId相同)的Checkpoint快照全部處理成功後獲取應用中所有實現了CheckpointListener的Operator并觸發執行notifyCheckpointComplete方法。觸發notifyCheckpointComplete方法時攜帶的checkpointId參數用來告訴Operator哪一輪Checkpoint已經完成。
1public interface CheckpointListener {
2  void notifyCheckpointComplete(long checkpointId) throws Exception;
3}

3.Flink Kafka端到端Exactly-once
Kafka是非常收歡迎的分布式消息系統,在Flink中它可以作為Source,同時也可以作為Sink。Kafka 0.11.0及以上版本提供了對事務的支持,這讓Flink應用搭載Kafka實現端到端的exactly-once成為了可能。下面我們就來深入了解提供了事務支持的Kafka是如何與Flink結合實現端到端exactly-once的。

本文忽略了Barrier機制,所以示例和圖中都以單線程為例。Barrier在《Flink Checkpoint原理》有較多讨論。

Flink Kafka Consumer

Kafka自身提供了可重複消費消息的能力,Flink結合Kafka的這個特性以及自身Checkpoint機制,得以實現Flink Kafka Consumer的容錯。
Flink Kafka Consumer是Flink應用從Kafka獲取數據流消息的一個實現。除了數據流獲取、數據發送下遊算子這些基本功能外它還提供了完善的容錯機制。這些特性依賴了其内部的一些組件以及内置的數據結構協同處理完成。這裡,我們先簡單了解這些組件和内置數據結構的職責,再結合Flink 運行時故障恢複時 兩個不同的處理時機來看一看它們之間是如何協同工作的。

  • KafkaTopic元數據:從Kafka消費數據的前提是需要知道消費哪個topic,這個topic有多少個partition。組件AbstractPartitionDiscoverer負責獲得指定topic的元數據信息,并将獲取到的topic元數據信息封裝成KafkaTopicPartition集合。
  • KafkaTopicPartition:KafkaTopicPartition結構用于記錄topic與partition的對應關系,内部定義了String topicint partition兩個主要屬性。假設topic A有2個分區,通過組件AbstractPartitionDiscoverer處理後将得到由兩個KafkaTopicPartition對象組成的集合:KafkaTopicPartition(topic:A,partition:0)KafkaTopicPartition(topic:A, partition:1)
  • Kafka數據消費:作為Flink Source,Flink Kafka Consumer最主要的職責就是能從Kafka中獲取數據,交給下遊處理。在Kafka Consumer中AbstractFetcher組件負責完成這部分功能。除此之外Fetcher還負責offset的提交、KafkaTopicPartitionState結構的數據維護。
  • KafkaTopicPartitionState:KafkaTopicPartitionState是一個非常核心的數據結構,基于内部的4個基本屬性,Flink Kafka Consumer維護了topic、partition、已消費offset、待提交offset的關聯關系。Flink Kafka Consumer的容錯機制依賴了這些數據。除了這4個基本屬性外KafkaTopicPartitionState還有兩個子類,一個是支持PunctuatedWatermark的實現,另一個是支持PeriodicWatermark的實現,這兩個子類在原有基礎上擴展了對水印的支持,我們這裡不做過多讨論。

clipboard.png

  • 狀态持久化:Flink Kafka Consumer的容錯性依靠的是狀态持久化,也可以稱為狀态快照。對于Flink Kafka Consumer來說,這個狀态持久化具體是對topic、partition、已消費offset的對應關系做持久化。
    在實現中,使用ListState<Tuple2<KafkaTopicPartition,Long>>定義了狀态存儲結構,在這裡Long表示的是offset類型,所以實際上就是使用KafkaTopicPartition和offset組成了一個對兒,再添加到狀态後端集合。
  • 狀态恢複:當狀态成功持久化後,一旦應用出現故障,就可以用最近持久化成功的快照恢複應用狀态。在實現中,狀态恢複時會将快照恢複到一個TreeMap結構中,其中key是KafkaTopicPartition,value是對應已消費的offset。恢複成功後,應用恢複到故障前Flink Kafka Consumer消費的offset,并繼續執行任務,就好像什麼都沒發生一樣。

運行時
我們假設Flink應用正常運行,Flink Kafka Consumer消費topic為Topic-ATopic-A隻有一個partition。在運行期間,主要做了這麼幾件事

  • Kafka數據消費:KafkaFetcher不斷的從Kafka消費數據,消費的數據會發送到下遊算子并在内部記錄已消費過的offset。下圖描述的是Flink Kafka Consumer從消費Kafka消息到将消息發送到下遊算子的一個處理過程。

clipboard.png

接下來我們再結合消息真正開始處理後,KafkaTopicPartitionState結構中的數據變化。

clipboard.png

可以看到,随着應用的運行,KafkaTopicPartitionState中的offset屬性值發生了變化,它記錄了已經發送到下遊算子消息在Kafka中的offset。在這裡由于消息P0-C已經發送到下遊算子,所以KafkaTopicPartitionState.offset變更為2。

  • 狀态快照處理:如果Flink應用開啟了Checkpoint,JobManager會定期觸發Checkpoint。FlinkKafkaConsumer實現了CheckpointedFunction,所以它具備快照狀态(snapshotState)的能力。在實現中,snapshotState具體幹了這麼兩件事

下圖描述當一輪Checkpoint開始時FlinkKafkaConsumer的處理過程。在例子中,FlinkKafkaConsumer已經将offset=3的P0-D消息發送到下遊,當checkpoint觸發時将topic=Topic-A;partition=0;offset=3作為最後的狀态持久化到外部存儲。

  • 将當前快照輪次(CheckpointId)與topic、partition、offset寫入到一個待提交offset的Map集合,其中key是CheckpointId。
  • FlinkKafkaConsumer當前運行狀态持久化,即将topic、partition、offset持久化。一旦出現故障,就可以根據最新持久化的快照進行恢複。

下圖描述當一輪Checkpoint開始時FlinkKafkaConsumer的處理過程。在例子中,FlinkKafkaConsumer已經将offset=3的P0-D消息發送到下遊,當checkpoint觸發時将topic=Topic-A;partition=0;offset=3作為最後的狀态持久化到外部存儲。

clipboard.png

  • 快照結束處理:當所有算子基于同一輪次快照處理結束後,會調用CheckpointListener.notifyCheckpointComplete(checkpointId)通知算子Checkpoint完成,參數checkpointId指明了本次通知是基于哪一輪Checkpoint。在FlinkKafkaConsumer的實現中,接到Checkpoint完成通知後會變更KafkaTopicPartitionState.commitedOffset屬性值。最後再将變更後的commitedOffset提交到Kafka
    brokers或Zookeeper。

在這個例子中,commitedOffset變更為4,因為在快照階段,将topic=Topic-A;partition=0;offset=3的狀态做了快照,在真正提交offset時是将快照的offset + 1作為結果提交的。「源代碼KafkaFetcher.java 207行doCommitInternalOffsetsToKafka方法」

clipboard.png

故障恢複

Flink應用崩潰後,開始進入恢複模式。假設Flink Kafka Consumer最後一次成功的快照狀态是topic=Topic-A;partition=0;offset=3,在恢複期間按照下面的先後順序執行處理。

  • 狀态初始化
    狀态初始化階段嘗試從狀态後端加載出可以用來恢複的狀态。它由CheckpointedFunction.initializeState接口定義。在FlinkKafkaConsumer的實現中,從狀态後端獲得快照并寫入到内部存儲結構TreeMap,其中key是由KafkaTopicPartition表示的topic與partition,value為offset。下圖描述的是故障恢複的第一個階段,從狀态後端獲得快照,并恢複到内部存儲。

clipboard.png

  • function初始化

    function初始化階段除了初始化OffsetCommitMode和partitionDiscoverer外,還會初始化一個Map結構,該結構用來存儲應用待消費信息。如果應用需要從快照恢複狀态,則從待恢複狀态中初始化這個Map結構。下圖是該階段從快照恢複的處理過程。

clipboard.png

function初始化階段兼容了正常啟動和狀态恢複時offset的初始化。對于正常啟動過程,StartupMode的設置決定待消費信息中的結果。該模式共有5種,默認為StartupMode.GROUP_OFFSETS

clipboard.png

  • 開始執行
    在該階段中,将KafkaFetcher初始化、初始化内部消費狀态、啟動消費線程等等,其目的是為了将FlinkKafkaConsumer運行起來,下圖描述了這個階段的處理流程

clipboard.png

這裡對圖中兩個步驟做個描述

  • 步驟3,使用狀态後端的快照結果topic=Topic-A;partition=0;offset=3初始化Flink Kafka
    Consumer内部維護的Kafka處理狀态。因為是恢複流程,所以這個内部維護的處理狀态也應該随着快照恢複。
  • 步驟4,在真正消費Kafka數據前(指調用KafkaConsumer.poll方法),使用Kafka提供的seek方法将offset重置到指定位置,而這個offset具體算法就是狀态後端offset 1。在例子中,消費Kafka數據前将offset重置為4,所以狀态恢複後KafkaConsumer是從offset=4位置開始消費。「源代碼KafkaConsumerThread.java
    428行」

總結

上述的3個步驟是恢複期間主要的處理流程,一旦恢複邏輯執行成功,後續處理流程與正常運行期間一緻。最後對FlinkKafkaConsumer用一句話做個總結。

「将offset提交權交給FlinkKafkaConsumer,其内部維護Kafka消費及提交的狀态。基于Kafka可重複消費能力并配合Checkpoint機制和狀态後端存儲能力,就能實現FlinkKafkaConsumer容錯性,即Source端的Exactly-once語義」。

Flink Kafka Producer

Flink Kafka Producer是Flink應用向Kafka寫出數據的一個實現。在Kafka 0.11.0及以上版本中提供了事務支持,這讓Flink搭載Kafka的事務特性可以輕松實現Sink端的Exactly-once語義。關于Kafka事務特性在《Kafka幂等與事務》中做了詳細讨論。

在Flink Kafka Producer中,有一個非常重要的組件FlinkKafkaInternalProducer,這個組件代理了Kafka客戶端org.apache.kafka.clients.producer.KafkaProducer,它為Flink Kafka Producer操作Kafka提供了強有力的支撐。在這個組件内部,除了代理方法外,還提供了一些關鍵操作。個人認為,Flink Kafka Sink能夠實現Exactly-once語義除了需要Kafka支持事務特性外,同時也離不開FlinkKafkaInternalProducer組件提供的支持,尤其是下面這些關鍵操作:

事務重置FlinkKafkaInternalProducer組件中最關鍵的處理當屬事務重置,事務重置由resumeTransaction方法實現「源代碼FlinkKafkaInternalProducer.java 144行」。由于Kafka客戶端未暴露針對事務操作的API,所以在這個方法内部,大量的使用了反射。方法中使用反射獲得KafkaProducer依賴的transactionManager對象,并将狀态後端快照的屬性值恢複到transactionManager對象中,這樣以達到讓Flink Kafka Producer應用恢複到重啟前的狀态。
下面我們結合Flink 運行時故障恢複 兩個不同的處理時機來了解Flink Kafka Producer内部如何工作。

運行時

我們假設Flink應用正常運行,Flink Kafka Producer正常接收上遊數據并寫到Topic-B的Topic中,Topic-B隻有一個partition。在運行期間,主要做以下幾件事:

  • 數據發送到Kafka

    上遊算子不斷的将數據Sink到FlinkKafkaProducerFlinkKafkaProducer接到數據後封裝ProducerRecord對象并調用Kafka客戶端KafkaProducer.send方法将ProducerRecord對象寫入緩沖「源代碼FlinkKafkaProducer.java 616行」。下圖是該階段的描述:

clipboard.png

  • 狀态快照處理 Flink 1.7及以上版本使用FlinkKafkaProducer作為Kafka
    Sink,它繼承抽象類TwoPhaseCommitSinkFunction,根據名字就能知道,這個抽象類主要實現兩階段提交。為了集成Flink Checkpoint機制,抽象類實現了CheckpointedFunctionCheckpointListener,因此它具備快照狀态(snapshotState)能力。狀态快照處理具體做了下面三件事:

①調用KafkaProducer客戶端flush方法,将緩沖區内全部記錄發送到Kafka,但不提交。這些記錄寫入到Topic-B,此時這些數據的事務隔離級别為UNCOMMITTED,也就是說如果有個服務消費Topic-B,并且設置的isolation.level=read_committed,那麼此時這個消費端還無法poll到flush的數據,因為這些數據尚未commit。什麼時候commit呢?在快照結束處理階段進行commit,後面會提到。

②将快照輪次與當前事務記錄到一個Map表示的待提交事務集合中,key是當前快照輪次的CheckpointId,value是由TransactionHolder表示的事務對象。TransactionHolder對象内部記錄了transactionalId、producerId、epoch以及Kafka客戶端kafkaProducer的引用。

③持久化當前事務處理狀态,也就是将當前處理的事務詳情存入狀态後端,供應用恢複時使用。

下圖是狀态快照處理階段處理過程

clipboard.png

  • 快照結束處理
    TwoPhaseCommitSinkFunction實現了CheckpointListener,應用中所有算子的快照處理成功後會收到基于某輪Checkpoint完成的通知。當FlinkKafkaProducer收到通知後,主要任務就是提交上一階段産生的事務,而具體要提交哪些事務是從上一階段生成的待提交事務集合中獲取的。

clipboard.png

圖中第4步執行成功後,flush到Kafka的數據從UNCOMMITTED變更為COMMITTED,這意味着此時消費端可以poll到這批數據了。

2PC(兩階段提交)理論的兩個階段分别對應了FlinkKafkaProducer的狀态快照處理階段和快照結束處理階段,前者是通過Kafka的事務初始化、事務開啟、flush等操作預提交事務,後者是通過Kafka的commit操作真正執行事務提交。

故障恢複

Flink應用崩潰後,FlinkKafkaProducer開始進入恢複模式。下圖為應用崩潰前的狀态描述:

clipboard.png

在恢複期間主要的處理在狀态初始化階段。當Flink任務重啟時會觸發狀态初始化,此時應用與Kafka已經斷開了連接。但在運行期間可能存在數據flush尚未提交的情況。

如果想重新提交這些數據需要從狀态後端恢複當時KafkaProducer持有的事務對象,具體一點就是恢複當時事務的transactionalId、producerId、epoch。這個時候就用到了FlinkKafkaInternalProducer組件中的事務重置,在狀态初始化時從狀态後端獲得這些事務信息,并重置到當前KafkaProducer中,再執行commit操作。這樣就可以恢複任務重啟前的狀态,Topic-B的消費端依然可以poll到應用恢複後提交的數據。

需要注意的是:如果這個重置并提交的動作失敗了,可能會造成數據丢失。下圖描述的是狀态初始化階段的處理流程:

clipboard.png

總結

FlinkKafkaProducer故障恢複期間,狀态初始化是比較重要的處理階段。這個階段在Kafka事務特性的強有力支撐下,實現了事務狀态的恢複,并且使得狀态存儲占用空間最小。依賴Flink提供的TwoPhaseCommitSinkFunction實現類,我們自己也可以對Sink做更多的擴展。

本文作者:TalkingData 史天舒

你可能感興趣的

載入中...