如何獲取kafka某一topic中最新的offset？如果你在0.9版本以上，可以用最新的Consumer client 客戶端，有() / consumer.position() 可以用于得到當前最

如何獲取kafka某一topic中最新的offset？

如果你在0.9版本以上，可以用最新的Consumer client 客戶端，有() / consumer.position() 可以用于得到當前最新的offset： ${log.dirs}/replication-offset-checkpoint

如何刪除weblogicjmsdelayed狀態(tài)的消息？

消消氣~

你看到的“ Topic htmltrack_error is already marked for deletion”是正常的。Kafka刪除topic的過程實際上是異步的：命令行僅僅是提交一個刪除申請給到controller，并標記該topic為“待刪除”狀態(tài)。其實在Kafka服務器端真正的刪除邏輯應該已經(jīng)開啟，正常情況下您需要多等待一段時間topic自然會被成功刪除。

spark streaming集群如何做容災處理？

spark streaming集群做容災處理spark streaming有兩種容錯機制：spark自帶的checkpoint，使用Kafka direct自行維護offset。

第一種方法中，checkpoint是spark streaming自帶的一種檢查點機制，可以通過一些特殊配置把輸入數(shù)據(jù)和計算過程中的數(shù)據(jù)存儲在可靠的文件系統(tǒng)中（比如hdfs或s3）。

它可以保存計算過程中的狀態(tài)，在發(fā)生失敗時可以控制回溯到什么程度，而不用重新進行計算。驅(qū)動器容錯，在驅(qū)動器崩潰重啟后，控制從什么位置繼續(xù)讀取數(shù)據(jù)。

checkpoint有三種應用方法元數(shù)據(jù)信息，主要包括：streaming應用程序的配置計算過程中一系列Dstream操作沒有完成的批處理，在運行隊列中的批處理但是沒有完成；消費數(shù)據(jù)的偏移量；編譯后的執(zhí)行程序（jar文件）序列化后的二進制文件。

第二種方法基于接收者的，采用push，由kafka的topic將數(shù)據(jù)推向spark，使用API：使用的是kafka高級消費者API

效果：kafka將數(shù)據(jù)推到spark執(zhí)行節(jié)點中并儲存起來，然后由Spark Streaming啟動作業(yè)來處理這些程序。

效果上，這種不是使用接收器接收，而是saprk每次拉取數(shù)據(jù)先去kafka中獲取上一次拉取的偏移量。根據(jù)偏移量獲取數(shù)據(jù)后，再進行處理。

筆者推薦第二種處理方法，自行維護offset要比spark自帶的checkpoint更好一些，更加靈活、安全，容災處理的能原力會更強。