基于Flink的MQ-Hive實時數(shù)據(jù)集成如何實現(xiàn)字節(jié)跳動

這篇文章主要介紹基于Flink的MQ-Hive實時數(shù)據(jù)集成如何實現(xiàn)字節(jié)跳動，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

在數(shù)據(jù)中臺建設過程中，一個典型的數(shù)據(jù)集成場景是將 MQ (Message Queue，例如 Kafka、RocketMQ 等)的數(shù)據(jù)導入到 Hive 中，以供下游數(shù)倉建設以及指標統(tǒng)計。由于 MQ-Hive 是數(shù)倉建設第一層，因此對數(shù)據(jù)的準確性以及實時性要求比較高。

字節(jié)跳動內已有解決方案如下圖所示，主要分了兩個步驟：

1. 通過 Dump 服務將 MQ 的數(shù)據(jù)寫入到 HDFS 文件

2. 再通過 Batch ETL 將 HDFS 數(shù)據(jù)導入到 Hive 中，并添加 Hive 分區(qū)

 

痛點

1. 任務鏈較長，原始數(shù)據(jù)需要經過多次轉換最終才能進入 Hive
2. 實時性比較差，Dump Service、Batch ETL 延遲都會導致最終數(shù)據(jù)產出延遲
3. 存儲、計算開銷大，MQ 數(shù)據(jù)重復存儲和計算
4. 基于原生 Java 打造，數(shù)據(jù)流量持續(xù)增長后，存在單點故障和機器負載不均衡等問題
5. 運維成本較高，架構上無法復用公司內 Hadoop/Flink/Yarn 等現(xiàn)有基礎設施
6. 不支持異地容災

優(yōu)勢

1. 基于流式引擎 Flink 開發(fā)，支持 Exactly Once 語義
2. 實時性更高，MQ 數(shù)據(jù)直接進入 Hive，無中間計算環(huán)節(jié)
3. 減少中間存儲，整個流程數(shù)據(jù)只會落地一次
4. 支撐 Yarn 部署模式，方便用戶遷移
5. 資源管理彈性，方便擴容以及運維
6. 支持雙機房容災

整體架構

1. DTS Source 接入不同 MQ 數(shù)據(jù)源，支持 Kafka、RocketMQ 等
2. DTS Sink 將數(shù)據(jù)輸出到目標數(shù)據(jù)源，支持 HDFS、Hive 等
3. DTS Core 貫穿整個數(shù)據(jù)同步流程，通過 Source 讀取源端數(shù)據(jù)，經過 DTS Framework 處理，最后通過 Sink 將數(shù)據(jù)輸出到目標端。
4. DTS Framework 集成類型系統(tǒng)、文件切分、Exactly Once、任務信息采集、事件時間、臟數(shù)據(jù)收集等核心功能
5. 支持 Yarn 部署模式，資源調度、管理比較彈性

 

Exactly Once

1. 數(shù)據(jù)寫入時，Source 端從上游 MQ 拉取數(shù)據(jù)并發(fā)送到 Sink 端；Sink 端將數(shù)據(jù)寫入到臨時目錄中
2. Checkpoint Snapshot 階段，Source 端將 MQ Offset 保存到 State 中；Sink 端關閉寫入的文件句柄，并保存當前 Checkpoint ID 到 State 中；
3. Checkpoint Complete 階段，Source 端 Commit MQ Offset；Sink 端將臨時目錄中的數(shù)據(jù)移動到正式目錄下
4. Checkpoint Recover 階段，加載最新一次成功的 Checkpoint 目錄并恢復 State 信息，其中 Source 端將 State 中保存的 MQ Offset 作為起始位置；Sink 端恢復最新一次成功的 Checkpoint ID，并將臨時目錄的數(shù)據(jù)移動到正式目錄下

■ 實現(xiàn)優(yōu)化

1. Sink 端臨時目錄為{dump_path}/{next_cp_id}，這里 next_cp_id 的定義是當前最新的 cp_id + 1
2. Checkpoint Snapshot 階段，Sink 端保存當前最新 cp_id 到 State，同時更新 next_cp_id 為 cp_id + 1
3. Checkpoint Complete 階段，Sink 端將臨時目錄中所有小于等于當前 cp_id 的數(shù)據(jù)移動到正式目錄下
4. Checkpoint Recover 階段，Sink 端恢復最新一次成功的 cp_id，并將臨時目錄中小于等于當前 cp_id 的數(shù)據(jù)移動到正式目錄下

類型系統(tǒng)

1. 在 Source 端，將源數(shù)據(jù)類型，統(tǒng)一轉成系統(tǒng)內部的 DTS 類型
2. 在 Sink 端，將系統(tǒng)內部的 DTS 類型轉換成目標數(shù)據(jù)源類型
3. 其中 DTS 類型系統(tǒng)支持不同類型間的相互轉換，比如 String 類型與 Date 類型的相互轉換

 

Rolling Policy

■ 優(yōu)化策略

1. RowFormat：基于單條寫入，支持按照 Offset 進行 HDFS Truncate 操作，例如 Text 格式
2. BulkFormat：基于 Block 寫入，不支持 HDFS Truncate 操作，例如 Parquet、ORC 格式

容錯處理

1. Flink 計算引擎升級，需要重啟任務
2. 上游數(shù)據(jù)增加，需要調整任務并發(fā)度
3. Task Failover

■ 并發(fā)度調整

■ Task Failover

異地容災

■ 背景

■ 容災組件

1. MQ 需要支持多機房部署，當主機房故障時，能將 Leader 切換到備機房，以供下游消費
2. Yarn 集群在主機房、備機房都有部署，以便 Flink Job 遷移
3. 下游 HDFS 需要支持多機房部署，當主機房故障時，能將 Master 切換到備機房
4. Flink Job 運行在 Yarn 上，同時任務 State Backend 保存到 HDFS，通過 HDFS 的多機房支持保障 State Backend 的多機房

 

   

 

■ 容災過程

1. 正常情況下，MQ Leader 以及 HDFS Master 部署在主機房，并將數(shù)據(jù)同步到備機房。同時 Flink Job 運行在主機房，并將任務 State 寫入到 HDFS 中，注意 State 也是多機房部署模式
2. 災難情況下，MQ Leader 以及 HDFS Master 從主機房遷移到備災機房，同時 Flink Job 也遷移到備災機房，并通過 State 恢復災難前的 Offset 信息，以提供 Exactly Once 語義

 

 

事件時間歸檔

■ 背景

■  全局最小歸檔時間

 

■ 亂序處理

1. 事件時間大于全局最小歸檔時間
2. 事件時間大于分區(qū)最小歸檔時間

 

 

 

Hive 分區(qū)生成

■ 原理

1. 在 Sink 端，對于每個 Task 保存當前最小處理時間，需要滿足單調遞增的特性
2. 在 Checkpoint Complete 時，Task 上報最小處理時間到 JM 端
3. JM 拿到所有 Task 的最小處理時間后，可以得到全局最小處理時間，并以此作為 Hive 分區(qū)的最小就緒時間
4. 當最小就緒時間更新時，可判斷是否添加 Hive 分區(qū)

 

 

■ 動態(tài)分區(qū)

 

Messenger

■ 元信息采集

 

 

■ 臟數(shù)據(jù)收集

■ 大盤監(jiān)控

 

以上是“基于Flink的MQ-Hive實時數(shù)據(jù)集成如何實現(xiàn)字節(jié)跳動”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注億速云行業(yè)資訊頻道！