上篇文章，討論了在spark streaming中管理消費kafka的偏移量的方式，本篇就接著聊聊上次說升級失敗的案例。

事情發生一個月前，由於當時我們想提高spark streaming程式的並行處理效能，於是需要增加kafka分割槽個數，，這裡需要說下，在新版本spark streaming和kafka的整合中，按照官網的建議 spark streaming的executors的數量要和kafka的partition的個數保持相等，這樣每一個executor處理一個kafka partition的資料，效率是最高的。如果executors的數量大於kafka的分割槽個數，其實多餘的executors相當於是不會處理任何資料，這部分的程序其實是白白浪費效能。

如果executor的個數小於kafka partition的個數，那麼其實有一些executors程序是需要處理多個partition分割槽的資料的，所以官網建議spark executors的程序數和kafka partition的個數要保持一致。

那麼問題來了，如果想要提高spark streaming的並行處理效能，只能增加kafka的分割槽了，給kafka增加分割槽比較容易，直接執行一個命令即可，不過這裡需要注意，kafka的分割槽只能增加不能減少，所以新增分割槽要考慮到底多少個才合適。

接下來我們便增加了kafka分割槽的數量，同時修改了spark streaming的executors的個數和kafka的分割槽個數一一對應，然後就啟動了流程式，結果出現了比較詭異的問題，表現如下：

造幾條測試資料打入kafka中，發現程式總是隻能處理其中的一部分資料，而每次總有一些資料丟失。按理說程式碼沒有任何改動，只是增加kafka的分割槽和spark streaming的executors的個數，應該不會出現問題才對，於是又重新測了原來的舊分割槽和程式，發現沒有問題，經過對比發現問題只會出現在kafka新增分割槽後，然後出現這種丟資料的情況。然後和運維同學一起看了新增的kafka的分割槽的磁碟目錄是否有資料落入，經查詢發現新的分割槽確實已經有資料進入了，這就很奇怪了丟的資料到底是怎麼丟的？

最後我又檢查了我們自己儲存的kafka的offset，發現裡面的偏移量竟然沒有新增kafka的分割槽的偏移量，至此，終於找到問題所在，也就是說，如果沒有新增分割槽的偏移量，那麼程式執行時是不會處理新增分割槽的資料，而我們新增的分割槽確確實實有資料落入了，這就是為啥前面說的詭異的丟失資料的原因，其實是因為新增kafka的分割槽的資料程式並沒有處理過而這個原因正是我們的自己儲存offset中沒有記錄新增分割槽的偏移量。

問題找到了，那麼如何修復線上丟失的資料呢？

當時想了一個比較笨的方法，因為我們的kafka線上預設是保留7天的資料，舊分割槽的資料已經處理過，就是新增的分割槽資料沒有處理，所以我們刪除了已經處理過的舊的分割槽的資料，然後在業務流量底峰時期，重新啟了流程式，讓其從最早的資料開始消費處理，這樣以來因為舊的分割槽被刪除，只有新分割槽有資料，所以相當於是把丟失的那部分資料給修復了。修復完成後，又把程式停止，然後配置從最新的偏移量開始處理，這樣偏移量裡面就能識別到新增的分割槽，然後就繼續正常處理即可。

注意這裡面的刪除kafka舊分割槽的資料，是一個比較危險的操作，它要求kafka的節點需要全部重啟才能生效，所以除非特殊情況，不要使用這麼危險的方式。

後來，仔細分析了我們使用的一個開源程式管理offset的原始碼，發現這個程式有一點bug，沒有考慮到kafka新增分割槽的情況，也就是說如果你的kafka分割槽增加了，你的程式在重啟後是識別不到新增的分割槽的，所以如果新增的分割槽還有資料進入，那麼你的程式一定會丟資料，因為擴充套件kafka分割槽這個操作，並不常見，所以這個bug比較難易觸發。

知道原因後，解決起來比較容易了，就是每次啟動流程式前，對比一下當前我們自己儲存的kafka的分割槽的個數和從zookeeper裡面的存的topic的分割槽個數是否一致，如果不一致，就把新增的分割槽給新增到我們自己儲存的資訊中，併發偏移量初始化成0，這樣以來在程式啟動後，就會自動識別新增分割槽的資料。

所以，回過頭來看上面的那個問題，最簡單優雅的解決方法就是，直接手動修改我們自己的儲存的kafka的分割槽偏移量資訊，把新增的分割槽給加入進去，然後重啟流程式即可。

這個案例也就是我上篇文章所說的第三個場景的case，如果是自己手動管理kafka的offset一定要注意相容新增分割槽後的這種情況，否則程式可能會出現丟失資料的問題。