目錄
- ETL 系統核心特征
- 資料重跑及其優化
- 重跑的場景
- 重跑的方式
- 重跑的優化
- 自動水平擴展
- 資料重跑及其優化
- 參考鏈接
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
美圖離線ETL實踐 - 掘金 (juejin.cn)
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/544236.html
標籤：其他