前面的文章已經介紹了在spark streaming整合kafka時，如何處理其偏移量的問題，由於spark streaming自帶的checkpoint弊端非常明顯，所以一些對資料一致性要求比較高的專案裡面，不建議採用其自帶的checkpoint來做故障恢復。

在spark streaming1.3之後的版本支援direct kafka stream，這種策略更加完善，放棄了原來使用Kafka的高階API自動儲存資料的偏移量，之後的版本採用Simple API也就是更加偏底層的api，我們既可以用checkpoint來容災，也可以通過低階api來獲取偏移量自己管理偏移量，這樣以來無論是程序升級，還是故障重啟，在框架端都可以做到Exact One準確一次的語義。

本篇文章，會再介紹下，如何手動管理kafka的offset，並給出具體的程式碼加以分析：

版本：

apache spark streaming2.1

apache kafka 0.9.0.0

手動管理offset的注意點：

（1）第一次專案啟動的時候，因為zk裡面沒有偏移量，所以使用KafkaUtils直接建立InputStream，預設是從最新的偏移量開始消費，這一點可以控制。

（2）如果非第一次啟動，zk裡面已經存在偏移量，所以我們讀取zk的偏移量，並把它傳入到KafkaUtils中，從上次結束時的偏移量開始消費處理。

（3）在foreachRDD裡面，對每一個批次的資料處理之後，再次更新存在zk裡面的偏移量

注意上面的3個步驟，1和2只會載入一次，第3個步驟是每個批次裡面都會執行一次。

下面看第一和第二個步驟的核心程式碼：

/****
    *
    * @param ssc  StreamingContext
    * @param kafkaParams  配置kafka的引數
    * @param zkClient  zk連線的client
    * @param zkOffsetPath zk裡面偏移量的路徑
    * @param topics     需要處理的topic
    * @return   InputDStream[(String, String)] 返回輸入流
 

    */
  def createKafkaStream(ssc: StreamingContext,
                        kafkaParams: Map[String, String],
                        zkClient: ZkClient,
                        zkOffsetPath: String,
                        topics: Set[String]): InputDStream[(String, String)]={
    //目前僅支援一個topic的偏移量處理，讀取zk裡面偏移量字串
    val zkOffsetData=KafkaOffsetManager.readOffsets(zkClient,zkOffsetPath,topics.last)

    val kafkaStream = zkOffsetData match {
      case None =>  //如果從zk裡面沒有讀到偏移量，就說明是系統第一次啟動
        log.info("系統第一次啟動，沒有讀取到偏移量，預設就最新的offset開始消費")
        //使用最新的偏移量建立DirectStream
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
      case Some(lastStopOffset) =>
        log.info("從zk中讀取到偏移量，從上次的偏移量開始消費資料......")
        val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.key, mmd.message)
        //使用上次停止時候的偏移量建立DirectStream
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, lastStopOffset, messageHandler)
    }
    kafkaStream//返回建立的kafkaStream
  }

主要是針對第一次啟動，和非首次啟動做了不同的處理。

然後看下第三個步驟的程式碼：

/****
    * 儲存每個批次的rdd的offset到zk中
    * @param zkClient zk連線的client
    * @param zkOffsetPath   偏移量路徑
    * @param rdd     每個批次的rdd
    */
  def saveOffsets(zkClient: ZkClient, zkOffsetPath: String, rdd: RDD[_]): Unit = {
    //轉換rdd為Array[OffsetRange]
    val offsetsRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    //轉換每個OffsetRange為儲存到zk時的字串格式 :  分割槽序號1:偏移量1,分割槽序號2:偏移量2,......
    val offsetsRangesStr = offsetsRanges.map(offsetRange => s"${offsetRange.partition}:${offsetRange.untilOffset}").mkString(",")
    log.debug(" 儲存的偏移量：  "+offsetsRangesStr)
    //將最終的字串結果儲存到zk裡面
    ZkUtils.updatePersistentPath(zkClient, zkOffsetPath, offsetsRangesStr)
  }

主要是更新每個批次的偏移量到zk中。

例子已經上傳到github中，有興趣的同學可以參考這個連結：

後續文章會聊一下為了升級應用如何優雅的關閉的流程式，以及在kafka擴充套件分割槽時，上面的程式如何自動相容。