1. 背景

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

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

初版架構問題

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

2. 架構優化升級

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

2.1 Canal

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

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

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

2.2 Apache Hudi

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

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

3. Hudi 實踐經驗分享

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

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

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

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

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

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

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

hoodie.combine.before.insert

hoodie.parquet.small.file.limit

hoodie.merge.allow.duplicate.on.inserts

其中：hoodie.combine.before.insert 決定是否對同一批次的數據按 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 的數據進行合并。此時有概率發生去重的情況 （如果相同 recordKey 的數據寫入同一文件中）；如果 hoodie.parquet.small.file.limit > 0 并且 hoodie.merge.allow.duplicate.on.inserts 為 true，那么在小文件合并的時候，不會處理相同 recordKey 的數據

4. 總結

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

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