1. 背景

OnZoom是Zoom新產品，是基于Zoom Meeting的一個獨一無二的在線活動平臺和市場。作為Zoom統(tǒng)一通信平臺的延伸，OnZoom是一個綜合性解決方案，為付費的Zoom用戶提供創(chuàng)建、主持和盈利的活動，如健身課、音樂會、站立表演或即興表演，以及Zoom會議平臺上的音樂課程。

在OnZoom data platform中，source數(shù)據主要分為MySQL DB數(shù)據和Log數(shù)據。 其中Kafka數(shù)據通過Spark Streaming job實時消費,MySQL數(shù)據通過Spark Batch job定時同步, 將source數(shù)據Sink到AWS S3。之后定時調度Spark Batch Job進行數(shù)倉開發(fā)。最終按照實際業(yè)務需求或使用場景將數(shù)據Sink到合適的存儲。

初版架構問題

• MySQL通過sql方式獲取數(shù)據并同步到S3是離線處理，并且某些場景下(比如物理刪除)只能每次全量同步
• Spark Streaming job sink到S3需要處理小文件問題
• 默認S3存儲方式不支持CDC(Change Data Capture),所以只支持離線數(shù)倉
• 因為安全要求，有時需求刪除或更新某個客戶數(shù)據時，只能全量(或指定分區(qū))計算并overwrite。性能較差

2. 架構優(yōu)化升級

基于以上問題，我們在進行大量技術調研選型及POC之后，我們主要做了如下2部分大的架構優(yōu)化升級。

2.1 Canal

MySQL Binlog即二進制日志,它記錄了MySQL所有表結構和表數(shù)據變更。

Cannal基于MySQL Binlog日志解析，提供增量數(shù)據訂閱和消費,將數(shù)據Sink到Kafka實現(xiàn)CDC。

后續(xù)使用Spark Streaming job實時消費Binlog就能解決上述問題1的時效性以及物理刪除等問題。

2.2 Apache Hudi

我們需要有一種能夠兼容S3存儲之后，既支持大量數(shù)據的批處理又支持增加數(shù)據的流處理的數(shù)據湖解決方案。最終我們選擇Hudi作為我們數(shù)據湖架構方案，主要原因如下:

• Hudi通過維護索引支持高效的記錄級別的增刪改
• Hudi維護了一條包含在不同的即時時間（instant time）對數(shù)據集做的所有instant操作的timeline,可以獲取給定時間內的CDC數(shù)據(增量查詢)。也提供了基于最新文件的Raw Parquet 讀優(yōu)化查詢。從而實現(xiàn)流批一體架構而不是典型的Lambda架構。
• Hudi智能自動管理文件大小，而不用用戶干預就能解決小文件問題
• 支持S3存儲，支持Spark、Hive、Presto查詢引擎，入門成本較低只需引入對應Hudi package

3. Hudi 實踐經驗分享

Hudi upsert 時默認PAYLOAD_CLASS_OPT_KEY為OverwriteWithLatestAvroPayload，該方式upsert時會將所有字段都更新為當前傳入的DataFrame。但很多場景下可能只想更新其中某幾個字段，其他字段跟已有數(shù)據保持一致，此時需要將PAYLOAD_CLASS_OPT_KEY傳為OverwriteNonDefaultsWithLatestAvroPayload，將不需要更新的字段設為null。但該upsert方式也有一定限制，比如不能將某個值更新為null。

我們現(xiàn)在有實時同步數(shù)據，離線rerun數(shù)據的場景，但當前使用的是Hudi 0.7.0版本，該版本還不支持多個job并發(fā)寫Hudi表。臨時方案是每次需要rerun數(shù)據的時候暫停實時任務，因為0.8.0版本已經支持并發(fā)寫，后續(xù)考慮升級。

一開始我們任務變更Hudi表數(shù)據時每次都默認同步hive元數(shù)據。但對于實時任務每次連接Hive Metastore更新元數(shù)據很浪費資源，因為大部分操作只涉及到數(shù)據變更而不涉及表結構或者分區(qū)變動。所以我們后來將實時任務關閉同步hive元數(shù)據，在需要更新元數(shù)據時另外再執(zhí)行hudi-hive-sync-bundle-*.jar來同步。

Hudi增量查詢語義是返回給定時間內所有的變更數(shù)據，所以會在timeline在里查找歷史所有commits文件。但歷史commits文件會根據retainCommits參數(shù)被清理，所以如果給定時間跨度較大時可能會獲取不到完整的變更數(shù)據。如果只關心數(shù)據的最終狀態(tài)，可以根據_hoodie_commit_time來過濾獲取增量數(shù)據。

Hudi默認spark分區(qū)并行度withParallelism為1500，需要根據實際的輸入數(shù)據大小調整合適的shuffle并行度。(對應參數(shù)為 hoodie.[insert|upsert|bulkinsert].shuffle.parallelism)

Hudi基于parquet列式存儲，支持向后兼容的schema evolution,但只支持新的DataFrame增加字段的schema變更，預計在在 0.10 版本實現(xiàn) full schema evolution。如果有刪除或重命名字段的需求，只能overwrite。另外增加字段也可能導致hive sync metadata失敗，需要先在hive執(zhí)行drop table。

Hudi Insert 對 recordKey 相同的數(shù)據，根據不同的參數(shù)有不同的處理情況，決定性的參數(shù)包括以下三個:

hoodie.combine.before.insert

hoodie.parquet.small.file.limit

hoodie.merge.allow.duplicate.on.inserts

其中：hoodie.combine.before.insert 決定是否對同一批次的數(shù)據按 recordKey 進行合并，默認為 false；hoodie.parquet.small.file.limit 和hoodie.merge.allow.duplicate.on.inserts 控制小文件合并閾值和如何進行小文件合并。如果 hoodie.parquet.small.file.limit > 0 并且 hoodie.merge.allow.duplicate.on.inserts 為 false，那么在小文件合并的時候，會對相同 recordKey 的數(shù)據進行合并。此時有概率發(fā)生去重的情況 （如果相同 recordKey 的數(shù)據寫入同一文件中）；如果 hoodie.parquet.small.file.limit > 0 并且 hoodie.merge.allow.duplicate.on.inserts 為 true，那么在小文件合并的時候，不會處理相同 recordKey 的數(shù)據

4. 總結

我司基于Hudi實現(xiàn)流批一體數(shù)據湖架構上線生產環(huán)境已有半年多時間，在引入Hudi之后我們在以下各個方面都帶來了一定收益:

• 成本:?引入Hudi數(shù)據湖方案之后，實現(xiàn)了S3數(shù)據增量查詢和增量更新刪除，之前更新刪除方案只能全表overwrite。Hudi實現(xiàn)智能小文件合并，之前需要單獨任務去處理。在數(shù)據處理和存儲方面都節(jié)約了相應成本，預估節(jié)省1/4費用。
• 時效性:?所有ODS表已從T+1改造為Near Real Time。后續(xù)會建設更多實時表。
• 效率:?在插入及更新數(shù)據時,默認情況下，Hudi使用Bloom Index，該索引更適合單調遞增record key，相比于原始Spark Join，其速度最高可提高10倍。查詢數(shù)據時，借助Hudi提供的Clustering(將文件按照某些列進行聚簇，以重新布局，達到優(yōu)化查詢性能的效果)，Compaction(將基礎文件和增量日志文件進行合并，生成新版本列存文件)等服務，可將查詢性能提升50%+。