1. 背景

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

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

初版架構(gòu)問題

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

2. 架構(gòu)優(yōu)化升級(jí)

基于以上問題，我們?cè)谶M(jìn)行大量技術(shù)調(diào)研選型及POC之后，我們主要做了如下2部分大的架構(gòu)優(yōu)化升級(jí)。

2.1 Canal

MySQL Binlog即二進(jìn)制日志,它記錄了MySQL所有表結(jié)構(gòu)和表數(shù)據(jù)變更。

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

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

2.2 Apache Hudi

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

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

3. Hudi 實(shí)踐經(jīng)驗(yàn)分享

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

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

一開始我們?nèi)蝿?wù)變更Hudi表數(shù)據(jù)時(shí)每次都默認(rèn)同步hive元數(shù)據(jù)。但對(duì)于實(shí)時(shí)任務(wù)每次連接Hive Metastore更新元數(shù)據(jù)很浪費(fèi)資源，因?yàn)榇蟛糠植僮髦簧婕暗綌?shù)據(jù)變更而不涉及表結(jié)構(gòu)或者分區(qū)變動(dòng)。所以我們后來將實(shí)時(shí)任務(wù)關(guān)閉同步hive元數(shù)據(jù)，在需要更新元數(shù)據(jù)時(shí)另外再執(zhí)行hudi-hive-sync-bundle-*.jar來同步。

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

Hudi默認(rèn)spark分區(qū)并行度withParallelism為1500，需要根據(jù)實(shí)際的輸入數(shù)據(jù)大小調(diào)整合適的shuffle并行度。(對(duì)應(yīng)參數(shù)為 hoodie.[insert|upsert|bulkinsert].shuffle.parallelism)

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

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

hoodie.combine.before.insert

hoodie.parquet.small.file.limit

hoodie.merge.allow.duplicate.on.inserts

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

4. 總結(jié)

我司基于Hudi實(shí)現(xiàn)流批一體數(shù)據(jù)湖架構(gòu)上線生產(chǎn)環(huán)境已有半年多時(shí)間，在引入Hudi之后我們?cè)谝韵赂鱾€(gè)方面都帶來了一定收益:

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

以上就是OnZoom基于Apache Hudi的一體架構(gòu)實(shí)踐 的詳細(xì)內(nèi)容，更多關(guān)于OnZoom基于Apache Hudi架構(gòu)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論