數(shù)據(jù)湖 Iceberg 在小米的落地及實(shí)踐

1、數(shù)據(jù)湖IC E B E RG 在小米 的落地與實(shí)踐目錄 CO NT E NT01 Iceberg 技術(shù)簡(jiǎn)介02 Iceberg在小米 的應(yīng)用實(shí)踐03 基于 Iceberg 的 流批一體的探索04 未來(lái)規(guī)劃Iceberg技術(shù)簡(jiǎn)介01|Iceberg 簡(jiǎn)介Apache Iceberg is an open table format for huge analytic datasets.Iceberg adds tables to compute engines including Spark, Trino, PrestoDB, Flink and Hive using a high-perform

2、ance table format that works just like a SQL table.|Iceberg 簡(jiǎn)介存儲(chǔ)與計(jì)算分離計(jì)算引擎選擇更靈活屏蔽底層文件存儲(chǔ)細(xì) 節(jié)，對(duì)外暴露都是一 張Iceberg表|文件布局|Metadata(元數(shù)據(jù)文件)記錄了最新的快照信息和歷史快照 信息，以及最新的Schema信息。Snapshot(快照文件)每次事務(wù)提交都會(huì)生成一個(gè) Snapshot。記錄了本次提交新增的 清單文件和歷史清單文件列表 (Manifest List)。Manifest(清單文件)記錄了本次事務(wù)寫入的文件和分區(qū) 的對(duì)應(yīng)關(guān)系，以及字段統(tǒng)計(jì)信息(最 大值、最小值)。Data Fi

3、le(數(shù)據(jù)文件)實(shí)際寫入的數(shù)據(jù)文件，如Parquet、Avro等格式文件。事務(wù)性避免 臟數(shù)據(jù)讀寫 分離|隱式分區(qū)根據(jù)數(shù)據(jù)自動(dòng)推斷分區(qū)分區(qū)不和目錄強(qiáng)綁定靈活的分區(qū)變更|行級(jí)更新format version更新方式V1Copy On WriteV2Copy On Write、Merge On Read|Iceberg在小米的 應(yīng)用實(shí)踐02|數(shù)據(jù)湖在小米應(yīng)用現(xiàn)狀表數(shù)量4000+|數(shù)據(jù)量v1表數(shù)量 8PB+1000+v2表數(shù)量3000+ChangLog 數(shù)據(jù)入湖|ChangLog 數(shù)據(jù)入湖優(yōu)點(diǎn)近實(shí)時(shí) 分析|支持流式 消費(fèi)同步變更Schema成本低數(shù)據(jù)均勻分布所有分區(qū)都可能需要 Compaction隨著

4、數(shù)據(jù)量增加，分區(qū)數(shù)不變ChangLog 入湖分區(qū)的選擇-自增 Id 為主鍵Bucket 分區(qū)Truncate 分區(qū)|數(shù)據(jù)只寫入最近幾個(gè)分區(qū)只對(duì)最近幾個(gè)分區(qū) Compaction隨著數(shù)據(jù)量增加分區(qū)數(shù)遞增ChangeLog 數(shù)據(jù)入湖產(chǎn)品化|日志數(shù)據(jù)入湖隱式分區(qū)避免數(shù)據(jù)漂移問題隱式分區(qū)保證延遲數(shù)據(jù)正確分區(qū)Flink + Iceberg 事務(wù)性保證數(shù)據(jù)不丟不重 支持 Schema 同步變更|日志數(shù)據(jù)入湖產(chǎn)品化|Iceberg-governance 服務(wù)Compaction服務(wù)|合并小文件、merge delete filesExpire Snapshots服務(wù)過期 snapshotsOrphan Fi

5、les clean 服務(wù)清理孤兒文件Hive 升級(jí) Iceberg壓縮方式存儲(chǔ)格式Parquet + ZSTD 存儲(chǔ)節(jié)約UNCOMPRESSEDTEXT80%SNAPPYSequenceFile30%+SNAPPYParquet30%+GZIPParquet5%|在Compaction 中配置更高的 compression level 獲得更高的壓縮率Hive 升級(jí) Iceberg 產(chǎn)品化|基于 Iceberg 的 流批一體的探索03|Lambda 架構(gòu)實(shí)時(shí)鏈路提供時(shí)效性離線鏈路提供準(zhǔn)確性離線數(shù)據(jù)支持回溯離線數(shù)據(jù)可供OLAP查詢實(shí)時(shí)鏈路不支持OLAP 查詢實(shí)時(shí)鏈路回溯能力有限兩套存儲(chǔ)，存儲(chǔ)成本

6、高兩套代碼，開發(fā)維護(hù)成本高實(shí)時(shí)離線數(shù)據(jù)不一致|數(shù)據(jù)湖架構(gòu)Iceberg 存儲(chǔ)上統(tǒng)一Flink 計(jì)算引擎統(tǒng)一支持回溯支持 OLAP 查詢支持構(gòu)建變更流|為什么需要離線作業(yè)來(lái)修數(shù)據(jù)？Flink 狀態(tài)過期導(dǎo)致沒 Join 上Watermark 設(shè)置導(dǎo)致延遲數(shù)據(jù)丟失Lookup Join 完成后維表發(fā)生了變更|Overwrite VS Merge Into ?MERGE INTO prod.db.target t- a target table USING (SELECT .) s- the source updatesON t.id = s.id AND t.date=20220601- condi

7、tion to find updates for target rows WHEN MATCHED AND s.op = deleteTHEN DELETEWHEN MATCHED AND t.count IS NULL AND s.op = increment THEN UPDATE SET t.count = 0WHEN MATCHED AND s.op = increment THEN UPDATE SET t.count = t.count + 1w WHEN NOT MATCHEDTHEN INSERT *|Overwrite VS Merge Into ?Overwrite|分區(qū)覆

8、蓋語(yǔ)法簡(jiǎn)單性能好可能導(dǎo)致下游消費(fèi)波動(dòng)Merge Into增量更新語(yǔ)法復(fù)雜性能不如 overwrite下游只消費(fèi)變更數(shù)據(jù)Merge Into 增量同步數(shù)據(jù)Merge Into 更新 Iceberg, Flink 同步變更至下游|不同的分區(qū)帶來(lái)的問題ProcessTime 分區(qū)|實(shí)時(shí)總是寫入 T 分區(qū)離線修正 T-1 分區(qū)（Overwrite）實(shí)時(shí)離線處理的數(shù)據(jù) 無(wú)交集EventTime 分區(qū)實(shí)時(shí)寫入對(duì)應(yīng)分區(qū)離線修復(fù)歷史全量（ Merge Into ）實(shí)時(shí)離線處理的數(shù)據(jù) 存在交集Merge Into 的隔離級(jí)別帶來(lái)的問題Serializable isolation level與提交過程中其他已成功提交事務(wù)沖突則本次提交失敗離線作業(yè)失敗概率增加使用 Serializable isolation level , 實(shí)時(shí)只處理 T 分區(qū)的變更，T-N 分區(qū)變