Kafka 偏移量管理實現精確一次語義在Spark&Flink中的技術實踐-kafka商業應用實戰

時間  2019-12-18
標籤 kafka 偏移量 管理 實現 精確 一次 語義 spark&flink spark flink 技術 實踐 商業應用 實戰 欄目 Kafka 简体版
原文   原文鏈接

本套技術專欄是做者（秦凱新）平時工做的總結和昇華，並深度整理大量網上資源和專業書籍。經過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和集羣環境容量規劃等內容，請持續關注本套博客。QQ郵箱地址：1120746959@qq.com，若有任何學術交流，可隨時聯繫。java

1 Kafka 偏移量

1.1 Kafka 0.9 以前版本

這裏的偏移量是指 kafka consumer offset，在 Kafka 0.9 版本以前消費者偏移量默認被保存在 zookeeper 中（/consumers/<group.id>/offsets//），所以在初始化消費者的時候須要指定 zookeeper.hosts。apache

1.2 Kafka 0.9 以後版本

隨着 Kafka consumer 在實際場景的不斷應用，社區發現舊版本 consumer 把位移提交到 ZooKeeper 的作法並不合適。ZooKeeper 本質上只是一個協調服務組件，它並不適合做爲位移信息的存儲組件，畢竟頻繁高併發的讀/寫操做並非 ZooKeeper 擅長的事情。所以在 0.9 版本開始 consumer 將位移提交到 Kafka 的一個內部 topic（__consumer_offsets）中，該主題默認有 50 個分區，每一個分區 3 個副本。bootstrap

1.3 消息處理語義

  • at-most-once：最多一次，消息可能丟失，但不會被重複處理；
  • at-least-once：至少一次，消息不會丟失，但可能被處理屢次；
  • exactly-once：精確一次，消息必定會被處理且只會被處理一次。
  • 若 consumer 在消息消費以前就提交位移，那麼即可以實現 at-most-once，由於若 consumer 在提交位移與消息消費之間崩潰，則 consumer 重啓後會重新的 offset 位置開始消費，前面的那條消息就丟失了；相反地，
  • 若提交位移在消息消費以後，則可實現 at-least-once 語義。因爲 Kafka 沒有辦法保證消息處理成功與位移提交在同一個事務中完成，若消息消費成功了，也提交位移了，可是處理失敗了，所以 Kafka 默認提供的就是 at-least-once 的處理語義。

1.4 kafka offset 提交方式

  • 默認狀況下，consumer 是自動提交位移的，自動提交間隔是 5 秒，能夠經過設置 auto.commit.interval.ms 參數能夠控制自動提交的間隔。併發

    自動位移提交的優點是下降了用戶的開發成本使得用戶沒必要親自處理位移提交；劣勢是用戶不能細粒度地處理位移的提交，特別是在有較強的精確一次處理語義時（在這種狀況下，用戶可使用手動位移提交）。app

  • 手動位移提交就是用戶自行肯定消息什麼時候被真正處理完並能夠提交位移，用戶能夠確保只有消息被真正處理完成後再提交位移。若是使用自動位移提交則沒法保證這種時序性，所以在這種狀況下必須使用手動提交位移。異步

    設置使用手動提交位移很是簡單，僅僅須要在構建 KafkaConsumer 時設置 enable.auto.commit=false，而後調用 commitSync 或 commitAsync 方法便可。高併發

2 Spark 位移處理方式

2.1 auto.offset.reset設置思路

  • 對於 auto.offset.reset 我的推薦設置爲 earliest，初次運行的時候，因爲 __consumer_offsets 沒有相關偏移量信息，所以消息會從最開始的地方讀取；當第二次運行時，因爲 __consumer_offsets 已經存在消費的 offset 信息，所以會根據 __consumer_offsets 中記錄的偏移信息繼續讀取數據。工具

  • 此外，對於使用 zookeeper 管理偏移量而言，只須要刪除對應的節點，數據便可從頭讀取，也是很是方便。不過若是你但願從最新的地方讀取數據，不須要讀取舊消息，則能夠設置爲 latest。oop

    earilist:提交過度區，從Offset處讀取，若是沒有提交過offset,從頭讀取
   latest:提交過度區，從Offset處讀取，沒有從最新的數據開始讀取
   None：若是沒有提交offset,就會報錯，提交過offset,就從offset處讀取
複製代碼

2.2 訂閱 Kafka 主題

  • 基於正則訂閱主題，有如下好處：fetch

    無需羅列主題名，一兩個主題還好，若是有幾十個，羅列過於麻煩了；
  可實現動態訂閱的效果（新增的符合正則的主題也會被讀取）。

  stream = KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.SubscribePattern[String, String](Pattern.compile(topicStr), kafkaConf, customOffset))
複製代碼

  • LocationStrategies 分配分區策略，LocationStrategies：根據給定的主題和集羣地址建立consumer

    建立DStream，返回接收到的輸入數據
    LocationStrategies.PreferConsistent：持續的在全部Executor之間勻分配分區 (均勻分配，選中的每個Executor都會分配 partition)
    LocationStrategies.PreferBrokers: 若是executor和kafka brokers 在同一臺機器上，選擇該executor。
    LocationStrategies.PreferFixed: 若是機器不是均勻的狀況下，能夠指定特殊的hosts。固然若是不指定，採用 LocationStrategies.PreferConsistent模式
複製代碼

  • SparkStreaming 序列化問題

    在 driver 中使用到的變量或者對象無需序列化，傳遞到 exector 中的變量或者對象須要序列化。所以推薦的作法是，在 exector 中最好只處理數據的轉換，在 driver 中對處理的結果進行存儲等操做。

    stream.foreachRDD(rdd => {
    
    // driver 代碼運行區域
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    kafkaOffset.updateOffset(offsetRanges)
  
    // exector 代碼運行區域
    val resultRDD = rdd.map(xxxxxxxx)
    //endregion
  
    //對結果進行存儲
    resultRDD.saveToES(xxxxxx)
    kafkaOffset.commitOffset(offsetRanges)
  })
複製代碼

2.3 使用老式zookeeper手動管理位移代碼分析

  • Zookeeper 偏移量管理ZkKafkaOffset實現，藉助 zookeeper 管理工具能夠對任何一個節點的信息進行修改、刪除，若是但願從最開始讀取消息，則只須要刪除 zk 某個節點的數據便可。

    import org.I0Itec.zkclient.ZkClient
  import org.apache.kafka.clients.consumer.ConsumerRecord
  import org.apache.kafka.common.TopicPartition
  import org.apache.spark.SparkConf
  import org.apache.spark.streaming.StreamingContext
  import org.apache.spark.streaming.kafka010.OffsetRange
  
  import scala.collection.JavaConverters._
  
  class ZkKafkaOffset(getClient: () => ZkClient, getZkRoot : () => String) {
  
    // 定義爲 lazy 實現了懶漢式的單例模式，解決了序列化問題，方便使用 broadcast
    lazy val zkClient: ZkClient = getClient()
    lazy val zkRoot: String = getZkRoot()
  
    // offsetId = md5(groupId+join(topics))
    // 初始化偏移量的 zk 存儲路徑 zkRoot
    def initOffset(offsetId: String) : Unit = {
      if(!zkClient.exists(zkRoot)){
        zkClient.createPersistent(zkRoot, true)
      }
    }
  
    // 從 zkRoot 讀取偏移量信息
    def getOffset(): Map[TopicPartition, Long] = {
      val keys = zkClient.getChildren(zkRoot)
      var initOffsetMap: Map[TopicPartition, Long] = Map()
      if(!keys.isEmpty){
        for (k:String <- keys.asScala) {
          val ks = k.split("!")
          val value:Long = zkClient.readData(zkRoot + "/" + k)
          initOffsetMap += (new TopicPartition(ks(0), Integer.parseInt(ks(1))) -> value)
        }
      }
      initOffsetMap
    }
  
    // 根據單條消息，更新偏移量信息
    def updateOffset(consumeRecord: ConsumerRecord[String, String]): Boolean = {
      val path = zkRoot + "/" + consumeRecord.topic + "!" + consumeRecord.partition
      zkClient.writeData(path, consumeRecord.offset())
      true
    }
  
    // 消費消息前，批量更新偏移量信息
    def updateOffset(offsetRanges: Array[OffsetRange]): Boolean = {
      for (offset: OffsetRange <- offsetRanges) {
        val path = zkRoot + "/" + offset.topic + "!" + offset.partition
        if(!zkClient.exists(path)){
          zkClient.createPersistent(path, offset.fromOffset)
        }
        else{
          zkClient.writeData(path, offset.fromOffset)
        }
      }
      true
    }
  
    // 消費消息後，批量提交偏移量信息
    def commitOffset(offsetRanges: Array[OffsetRange]): Boolean = {
      for (offset: OffsetRange <- offsetRanges) {
        val path = zkRoot + "/" + offset.topic + "!" + offset.partition
        if(!zkClient.exists(path)){
          zkClient.createPersistent(path, offset.untilOffset)
        }
        else{
          zkClient.writeData(path, offset.untilOffset)
        }
      }
      true
    }
  
    def finalize(): Unit = {
      zkClient.close()
    }
  }
  
  object ZkKafkaOffset{
    def apply(cong: SparkConf, offsetId: String): ZkKafkaOffset = {
      val getClient = () =>{
        val zkHost = cong.get("kafka.zk.hosts", "127.0.0.1:2181")
        new ZkClient(zkHost, 30000)
      }
      val getZkRoot = () =>{
        val zkRoot = "/kafka/ss/offset/" + offsetId
        zkRoot
      }
      new ZkKafkaOffset(getClient, getZkRoot)
    }
  }
複製代碼

  • Spark Streaming 消費 Kafka 消息

    第一步：val customOffset: Map[TopicPartition, Long] = kafkaOffset.getOffset(ssc)
  第二步：stream = KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](topics, kafkaConf, customOffset))
  第三步：處理後，kafkaOffset.commitOffset(offsetRanges)

  import scala.collection.JavaConverters._
  
  object RtDataLoader {
    def main(args: Array[String]): Unit = {
      // 從配置文件讀取 kafka 配置信息
      val props = new Props("xxx.properties")
      val groupId = props.getStr("groupId", "")
      if(StrUtil.isBlank(groupId)){
        StaticLog.error("groupId is empty")
        return
      }
      val kfkServers = props.getStr("kfk_servers")
      if(StrUtil.isBlank(kfkServers)){
        StaticLog.error("bootstrap.servers is empty")
        return
      }
      val topicStr = props.getStr("topics")
      if(StrUtil.isBlank(kfkServers)){
        StaticLog.error("topics is empty")
        return
      }
  
      // KAFKA 配置設定
      val topics = topicStr.split(",")
      val kafkaConf = Map[String, Object](
        "bootstrap.servers" -> kfkServers,
        "key.deserializer" -> classOf[StringDeserializer],
        "value.deserializer" -> classOf[StringDeserializer],
        "group.id" -> groupId,
        "receive.buffer.bytes" -> (102400: java.lang.Integer),
        "max.partition.fetch.bytes" -> (5252880: java.lang.Integer),
        "auto.offset.reset" -> "earliest",
        "enable.auto.commit" -> (false: java.lang.Boolean)
      )
  
      val conf = new SparkConf().setAppName("ss-kafka").setIfMissing("spark.master", "local[2]")
  
      // streaming 相關配置
      conf.set("spark.streaming.stopGracefullyOnShutdown","true")
      conf.set("spark.streaming.backpressure.enabled","true")
      conf.set("spark.streaming.backpressure.initialRate","1000")
  
      // 設置 zookeeper 鏈接信息
      conf.set("kafka.zk.hosts", props.getStr("zk_hosts", "sky-01:2181"))
  
      // 建立 StreamingContext
      val sc = new SparkContext(conf)
      sc.setLogLevel("WARN")
      val ssc = new StreamingContext(sc, Seconds(5))
  
      // 根據 groupId 和 topics 獲取 offset
      val offsetId = SecureUtil.md5(groupId + topics.mkString(","))
      val kafkaOffset = ZkKafkaOffset(ssc.sparkContext.getConf, offsetId)
      kafkaOffset.initOffset(ssc, offsetId)
      val customOffset: Map[TopicPartition, Long] = kafkaOffset.getOffset(ssc)
  
      // 建立數據流
      var stream:InputDStream[ConsumerRecord[String, String]] = null
      if(topicStr.contains("*")) {
        StaticLog.warn("使用正則匹配讀取 kafka 主題：" + topicStr)
        stream = KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.SubscribePattern[String, String](Pattern.compile(topicStr), kafkaConf, customOffset))
      }
      else {
        StaticLog.warn("待讀取的 kafka 主題：" + topicStr)
        stream = KafkaUtils.createDirectStream[String, String](ssc,
          LocationStrategies.PreferConsistent,
          ConsumerStrategies.Subscribe[String, String](topics, kafkaConf, customOffset))
      }
  
      // 消費數據
      stream.foreachRDD(rdd => {
        // 消息消費前，更新 offset 信息
        val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        kafkaOffset.updateOffset(offsetRanges)
  
        //region 處理詳情數據
        StaticLog.info("開始處理 RDD 數據！")
        //endregion
  
        // 消息消費結束，提交 offset 信息
        kafkaOffset.commitOffset(offsetRanges)
      })
      ssc.start()
      ssc.awaitTermination()
    }
  }
複製代碼

3 Flink 位移處理方式

3.1 Flink 消費者精確到一次語義

  • setStartFromGroupOffsets()【默認消費策略】 默認讀取上次保存的offset信息 若是是應用第一次啓動，讀取不到上次的offset信息，則會根據這個參數auto.offset.reset的值來進行消費數據

  • setStartFromEarliest() 從最先的數據開始進行消費，忽略存儲的offset信息

  • setStartFromLatest() 從最新的數據進行消費，忽略存儲的offset信息

  • setStartFromSpecificOffsets(Map<KafkaTopicPartition, Long>) 從指定位置進行消費。

  • 當checkpoint機制開啓的時候，KafkaConsumer會按期把kafka的offset信息還有其餘operator的狀態信息一塊保存起來。當job失敗重啓的時候，Flink會從最近一次的checkpoint中進行恢復數據，從新消費kafka中的數據。

  • 爲了可以使用支持容錯的kafka Consumer，須要開啓checkpoint env.enableCheckpointing(5000); // 每5s checkpoint一次

  • Kafka Consumers Offset 自動提交有如下兩種方法來設置，能夠根據job是否開啓checkpoint來區分:

    (1) Flink Checkpoint關閉時： 能夠經過Kafka下面兩個Properties參數配置

    enable.auto.commit
  auto.commit.interval.ms
複製代碼

    (2) Checkpoint開啓時：當執行checkpoint的時候纔會保存offset，這樣保證了kafka的offset和checkpoint的狀態偏移量保持一致。能夠經過這個參數設置

    setCommitOffsetsOnCheckpoints(boolean)
複製代碼

    這個參數默認就是true。表示在checkpoint的時候提交offset, 此時，kafka中的自動提交機制就會被忽略。

    //獲取Flink的運行環境
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

      //checkpoint配置
      env.enableCheckpointing(5000);
      env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
      env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
      env.getCheckpointConfig().setCheckpointTimeout(60000);
      env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
      env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

      //設置statebackend
      env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop100:9000/flink/checkpoints",true));

      String topic = "kafkaConsumer";
      Properties prop = new Properties();
      prop.setProperty("bootstrap.servers","SparkMaster:9092");
      prop.setProperty("group.id","kafkaConsumerGroup");

      FlinkKafkaConsumer011<String> myConsumer = new FlinkKafkaConsumer011<>(topic, new SimpleStringSchema(), prop);

      myConsumer.setStartFromGroupOffsets();//默認消費策略
      myConsumer.setCommitOffsetsOnCheckpoints(true);
      DataStreamSource<String> text = env.addSource(myConsumer);

      text.print().setParallelism(1);

      env.execute("StreamingFromCollection");
複製代碼

  • Flink KafkaConsumer容許配置向 Kafka brokers（或者向Zookeeper）提交offset的行爲。須要注意的是，Flink Kafka Consumer並不依賴於這些提交回Kafka或Zookeeper的offset來保證容錯。這些被提交的offset只是意味着Flink將消費的狀態暴露在外以便於監控。

  • FlinkKafkaConsumer提供了一套健壯的機制保證了在高吞吐量的狀況下exactly-once的消費Kafka的數據，它的API的使用與配置也比較簡單，同時也便於監控。

  • barrier能夠理解爲checkpoint之間的分隔符，在它以前的data屬於前一個checkpoint，而在它以後的data屬於另外一個checkpoint。同時，barrier會由source(如FlinkKafkaConsumer)發起，並混在數據中，同數據同樣傳輸給下一級的operator，直到sink爲止。若是barrier已經被sink收到，那麼說明checkpoint已經完成了(這個checkpoint的狀態爲completed並被存到了state backend中)，它以前的數據已經被處理完畢並sink。

  • Flink異步記錄checkpoint的行爲是由咱們的來配置的，只有當咱們設置了enableCheckpointing()時，Flink纔會在checkpoint完成時(整個job的全部的operator都收到了這個checkpoint的barrier才意味這checkpoint完成，具體參考咱們對Flink checkpoint的介紹)將offset記錄起來並提交，這時候纔可以保證exactly-once。

3.2 Flink 生產者精確到一次語義

  • Kafka Producer的容錯-Kafka 0.9 and 0.10

    若是Flink開啓了checkpoint，針對FlinkKafkaProducer09和FlinkKafkaProducer010 能夠提供 at-least-once的語義，還須要配置下面兩個參數:
  setLogFailuresOnly(false)
  setFlushOnCheckpoint(true)
  注意：建議修改kafka 生產者的重試次數retries【這個參數的值默認是0】
複製代碼

  • Kafka Producer的容錯-Kafka 0.11，若是Flink開啓了checkpoint，針對FlinkKafkaProducer011 就能夠提供 exactly-once的語義,可是須要選擇具體的語義

    具體的語義設置方式 
  Semantic.NONE
  Semantic.AT_LEAST_ONCE【默認】
  Semantic.EXACTLY_ONCE

  checkpoint配置
  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  env.enableCheckpointing(5000);
  env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
  env.getCheckpointConfig().setCheckpointTimeout(60000);
  env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  
  //第一種解決方案，設置FlinkKafkaProducer011裏面的事務超時時間
  //設置事務超時時間
  //prop.setProperty("transaction.timeout.ms",60000*15+"");

  //第二種解決方案，設置kafka的最大事務超時時間,主要是kafka的配置文件設置。

  //FlinkKafkaProducer011<String> myProducer = new FlinkKafkaProducer011<>(brokerList, topic, new SimpleStringSchema());

  //使用僅一次語義的kafkaProducer
  FlinkKafkaProducer011<String> myProducer = new FlinkKafkaProducer011<>(topic, new KeyedSerializationSchemaWrapper<String>(new SimpleStringSchema()), prop, FlinkKafkaProducer011.Semantic.EXACTLY_ONCE);
  
  text.addSink(myProducer);
複製代碼

4 總結

本套技術專欄是做者（秦凱新）平時工做的總結和昇華，並深度整理大量網上資源和專業書籍。經過從真實商業環境抽取案例進行總結和分享，並給出商業應用的調優建議和集羣環境容量規劃等內容，請持續關注本套博客。QQ郵箱地址：1120746959@qq.com，若有任何學術交流，可隨時聯繫。

本文在這裏一直困擾我如何Kafka 偏移量管理實現精確一次語義，若是你有幸閱讀到本篇內容，且有最好的解決方案，望可以給我留言，期待更好的解決方案。

秦凱新 於浪潮

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程