ETL 系統核心特征 數據重跑及其優化 重跑的場景 | 場景 | 導致原因 | 影響 | | | | | | kafka consumer poll消息失敗 | 1. 網路問題；2. kafka broker 磁碟壞道，拉取消息一直失敗或其他 kafka 原因 | 導致一個或多個topic&part ...
目錄
ETL 系統核心特征
數據重跑及其優化
重跑的場景
| 場景 | 導致原因 | 影響 |
|---|---|---|
| kafka consumer poll消息失敗 | 1. 網路問題；2. kafka broker 磁碟壞道，拉取消息一直失敗或其他 kafka 原因 | 導致一個或多個topic&partition的消息未消費完整 |
| 硬體故障，機器重啟 | 磁碟滿、硬體故障等 | 機器宕機、重啟、yarn內部機制會重新在另外一個nodeManager節點重新分配宕機節點的mapper task，可能會造成數據重覆 |
| task killed | 1. yarn 主動 killed task : ①， mapper 初始化某個邏輯卡住，導致 mappper 超時;如：載入調試設備信息時 mysql 阻塞、載入 lP 庫一直Full GC 或者 mapper 獲取不到執行資源一直等待等； ② yarn . nodemanager .local-dirs and yarn. nodemanager . log-dirs 配置的磁碟使用率超過 90 % ResourceManager 標記那台 NodeManger 為bad ，把機器上跑的 container 都 kill了，導致 ETL 一個 mapper 被 kill ，然後起一個新的 mapper task ，但是因為 kill mapper task attmept 未正常關閉，導致文件租約 Iease 未安全釋放，後面起 Mapper 一直寫失敗； 2.S手工 killed task ①Hadoop 集群節點負載太高，一些 DataNode 響應慢或者進程值死無響應，導致HDFS 寫入一直超時失敗等;人工 killed | 這個 mapper 分配執行的所有 topic partition 消息未消費或者消息不完整（影響多個 topic ) ;mapper 執行慢，數據一直寫不進去（創建、讀寫 block 超時） : |
當用戶調用 application kill 時會經歷三個階段：
- kill SIGTERM（-15） pid；
- Sleep for 250ms；
- 3.kill SIGKILL（-9） pid 。
重跑的方式
| 重跑方式 | 優點 | 缺點 |
|---|---|---|
| 1.備份目標目錄的數據 2.將 kafka topic &partition 的 offset 重新消費一遍清洗落地； |
簡單（由於每個 et l執行消費的offset 都是自己記錄維護的．找到當前小時或者當天對應的最小以及最大offset 重新消費即可） | 依賴於 kafka 消息過期時間，過期後無法重新消費； 因為大部分時間分區是是按天的．目前重跑一天的數需要 4 ＋小時．故障後恢復非常慢； |
| 記錄未消費的offset信息，補跑未消費的offset即可 | 速度快（只處理未消費完整的topic&partitioner offset） | 1 ）依賴於 kafka 消息過期時間，過期後無法重新消費 2 ) MapTask 被 yarn killed 或者人為 killed ，會導致寫數據的 Writer 不正常關閉流，從而引發 hdfs 文件租約 Iease 未釋放、 block recovery 等問題，導致後面數據 append 失敗； 3 ) MapTask attempt 被 yarn killed 後會自動起另外一個 attemPt （目前來．無法避免，這是 yarn 的機制，需要改源碼），可能導致數據重覆寫入： 4 ) killed 或者機器宕機是無法記錄到。offset 消息到哪裡，只能每段小段時間記錄當前 offset ．當 MapTask 被 killed . offset 可能不是最新的，導致補跑時重覆寫了部分數據； |
| 1 ) ETL 小時任務，數據落地到臨時目錄．文件按小時劃分．支持覆蓋寫（保存一段時間） ; 2 ）當數據落地成功，檢測每個 toPic partition 的消費情況進行處理； 3 ）如果 topic partitlon 的消息消費不完整時，告警通知，手工重跑相關 toPic partition 的小時任務； 4 ）如果消費完整，將小時數據文件合併到倉庫的目標文件（每個 topic 、 partition 單獨一個文件） ; 5 ）如果合併過程失敗．告警通知，手工觸發重新將已經落地的小文件合併成一個目標倉庫文件； |
支持小時重跑、覆蓋寫 恢復速度較快 HDFS block 丟失、可以從小文件新恢復 |
1 ）流程比較複雜，多出一個數據合併 Merge 的步驟； 2 ）每小時的 ETL 會產生非常多的小文件間定期刪除；需要保留一段時 |
如下圖所示是第三種重跑方式的整體流程，ETL 是按照小時調度的，首先將數據按小時寫到臨時目錄中，如果消費失敗會告警通知並重跑消費當前小時。如果落地成功則合併到倉庫目錄的目標文件，合併失敗同樣會告警通知並人工重跑，將小文件合併成目標文件。
graph TB 開始 --> 小時ETL--write -->1{落地臨時目錄/etl/work/按小時區分}--落地成功-->2{到倉庫目錄的目標文件每個partition一個文件}--合併成功-->結束 1-->失敗消費不完整-->告警通知-->重跑消費當前小時即可-->小時ETL 2-->合併失敗-->告警通知-->根據每小時產生的小文件重新合併一個目標文件--合併成功-->結束重跑的優化
| 流程 | 按天分區的topic |
|---|---|
| 每小時05分調度的ETL正常跑完 | 1.包含之前的數據．如：昨天漏採集的日誌．直接在對應日期分區目錄下創建一個新的數據文件； 2.今天的日誌，直接在今天的日期分區下創建一個新的數據文件； 3.將當前 ETL 產生的小文件直接 append 到對應日期分區以及相同 partition 下的文件； |
| 一個小時內執行多次ETL | 1.因為目前落地的臨時文件都是按 partition ＋執行時間（yyyyMMddHH ）生成的，如： all _ 1 _ 2018031511 . avro ; 2.如果一個小時執行多次 ETL ，每次都會消費新的 offset ( etl _ 0ffsets 會存多條記錄），但是臨時目錄的數據文件的名稱相同，這時候通過 append 方式追加數據； 3.如果小時重跑時．會找到當前小時內 topic--partjtion 的最小的 beginoffset 、最大的 endoffset 重新從 kafka 消費，這樣保證了數據的完整性； |
| 小時ETL重跑（前提未進行merge） | 1 ）小時三跑，直接找到對應小時 topic--partition 的最小的 beginoffset 、最大的 endoffset 重新從 kafka 消費： 2 ）之前已經創建的臨時目錄的數據文件會以 create overwrite 形式重新寫入數據； 3 ）最後在將文件 append 到對應日期分區下相同 partition 的文件； |
| merge失敗（小時ETL產生的臨時數據文件已經是完整、可靠的） | 重新根據當天每小時 ETL 產生的臨時數據文件合併一個目標倉庫文件即可： |
| ETL消費消息不完整 | 1.消費不完整的 toPic 、 partition 信息會記錄．告警簡訊通知： 2 ）人工重跑對應 topic--partition 小時 ETL 即可（支持細化到 partition 的小時數據重跑，因為是 overwrite ) |
| MapTask attempt killed by yarn | 1 ) MapTask attempt 被 yarn 的機制 killed 後．另啟的 attempt 會 overwrite 之前的文件（排除一個小時執行多次 ETL 的情況．這時候需要人工重跑）．比較好的相容 hadoop 集群的故障; |
| NodeManager carsh by hardware fault | 由於機器硬體故障導致運行 ETL 的 MapTask 的 NodeManager 直接 crash :1 ）數據正在寫入臨時目錄，手工重跑對應 topic 一 partition 小時的 ETL 即可： 2 ）數據正在 merge ，手工重新 merge 對應 topic date 的數據； |
| 人為killed整個ETL job | 人工處理未執行的topic ：1.數據正在寫入臨時目錄，手工重跑對於topic-partition小時的ETL即可 2.數據正在 merge , 手工重新merge對應 topic--date 的數據： |
自動水平擴展
現在離線 Kafka-ETL 是每小時 05 分調度，每次調度的 ETL 都會獲取每個 topic&partition 當前最新、最大的 latest offset，同時與上個小時消費的截止 offset 組合成本地要消費的 kafkaEvent。由於每次獲取的 latest offset 是不可控的，有些情況下某些 topic&partition 的消息 offset 增長非常快，同時 kafka topic 的 partition 數量來不及調整，導致 ETL 消費處理延遲，影響下游的業務處理流程：
- 由於擴容、故障等原因需要補採集漏採集的數據或者歷史數據，這種情況下 topic&&partition 的消息 offset 增長非常快，僅僅依賴 kafka topic partiton 擴容是不靠譜的，補採集完後面還得刪除擴容的 partition；
- 周末高峰、節假日、6.18、雙十一等用戶流量高峰期，收集的用戶行為數據會比平時翻幾倍、幾十倍，但是同樣遇到來不及擴容 topic partition 個數、擴容後需要縮容的情況；
Kafka ETL 是否能自動水平擴展不強依賴於 kafka topic partition 的個數。如果某個 topic kafkaEvent 需要處理的數據過大，評估在合理時間範圍單個 mapper 能消費的最大的條數，再將 kafkaEvent 水平拆分成多個子 kafkaEvent，並分配到各個 mapper 中處理，這樣就避免單個 mapper 單次需要處理過大 kafkaEvent 而導致延遲，提高水平擴展能力。拆分的邏輯如下圖所示：
graph TB 開始 --> 根據ETLSchema選擇SplitStrategy-->IsRemaining("SplitStrategy是否配置") IsRemaining -->配置--> 生成對應的SplitStrategy-->根據SplitStrategy拆分KafkaEvent-->oneChosse("消息是否超過閾值") oneChosse -->no["否"]-->KafkaEvent不拆分-->結束 oneChosse -->yes["是"]-->twoChosse("消息是否超過閾值2倍")-->yes2["是"]-->根據倍數進行合理拆分-->結束 twoChosse-->no2["否"]-->KafkaEvent拆分為兩個-->結束 IsRemaining -->未配置-->KafkaEvent不拆分-->結束後續我們將針對以下兩點進行自動水平擴展的優化：
- 如果單個 mapper 處理的總消息數據比較大，將考慮擴容 mapper 個數並生成分片 split 進行負載均衡。
- 每種格式的消息處理速度不一樣，分配時可能出現一些 mapper 負擔比較重，將給每個格式配置一定的權重，根據消息條數、權重等結合一起分配 kafkaEvent。
參考鏈接
https://blog.csdn.net/javastart/article/details/113838240
