據(jù)湖實(shí)戰(zhàn)：選型、調(diào)優(yōu)與運(yùn)維)
1. 流批一體場景下的數(shù)據(jù)湖選型思考1.1 從Lambda架構(gòu)到流批一體的演進(jìn)邏輯做數(shù)據(jù)平臺(tái)的人都有一個(gè)共同的痛同一份業(yè)務(wù)邏輯離線鏈路寫一遍實(shí)時(shí)鏈路再寫一遍兩套代碼、兩套運(yùn)維、兩套口徑。早期大家用Lambda架構(gòu)扛著離線跑T1出報(bào)表實(shí)時(shí)鏈路單獨(dú)維護(hù)一套指標(biāo)時(shí)間一長兩條鏈路的數(shù)據(jù)對(duì)不上業(yè)務(wù)方找過來排查成本極高。后來大家開始琢磨Kappa架構(gòu)把所有計(jì)算都收斂到流上但流處理引擎對(duì)大規(guī)模歷史數(shù)據(jù)的回溯分析、批量修正能力又不夠成熟于是“流批一體”這個(gè)概念被反復(fù)提起。流批一體的核心訴求其實(shí)很樸素一套SQL、一份元數(shù)據(jù)、一份存儲(chǔ)既能處理無界的實(shí)時(shí)流也能處理有界的歷史批。要做到這一點(diǎn)計(jì)算引擎和存儲(chǔ)層必須解耦存儲(chǔ)層要能同時(shí)承接高頻小批量的流式寫入和大吞吐的批量讀取。這就是數(shù)據(jù)湖技術(shù)在這個(gè)階段被推到臺(tái)前的原因。在眾多數(shù)據(jù)湖方案里Iceberg、Hudi、Delta Lake是繞不開的三個(gè)選項(xiàng)。我們當(dāng)時(shí)做選型的時(shí)候核心考量維度有幾個(gè)schema演進(jìn)能力、分區(qū)裁剪效率、小文件治理機(jī)制、與計(jì)算引擎的耦合程度、以及社區(qū)活躍度。Hudi的upsert能力很強(qiáng)適合高頻更新的場景但它和Spark的綁定相對(duì)深一些Delta Lake在Databricks生態(tài)里體驗(yàn)很好但開源版本在Flink集成上的成熟度當(dāng)時(shí)還不夠。Iceberg的設(shè)計(jì)哲學(xué)是“表格式規(guī)范優(yōu)先”它把表的元數(shù)據(jù)管理抽象得非常干凈計(jì)算引擎只需要按照規(guī)范去讀寫就行這種解耦特性讓它天然適合做流批一體的存儲(chǔ)底座。1.2 為什么是Flink加Iceberg的組合Flink在流處理上的優(yōu)勢不用多說真正的流式語義、事件時(shí)間處理、精確一次的狀態(tài)一致性這些都是它的看家本領(lǐng)。但Flink早期在批處理上的表現(xiàn)只能算“能用”直到1.12之后批流一體的SQL層逐漸成熟Flink才真正具備了統(tǒng)一計(jì)算引擎的底氣。Iceberg這邊它的核心抽象是表格式Table Format不是存儲(chǔ)引擎也不是文件格式。它定義了一套元數(shù)據(jù)規(guī)范包括snapshot、manifest、manifest list這些概念把“哪些文件屬于這張表”“這些文件在某個(gè)時(shí)間點(diǎn)的狀態(tài)是什么”管理得清清楚楚。這種設(shè)計(jì)帶來的好處是寫入方和讀取方完全解耦寫入方只管生成新的snapshot讀取方根據(jù)snapshot做快照隔離讀取互不干擾。Flink加Iceberg的組合本質(zhì)上是用Flink做統(tǒng)一的寫入和計(jì)算入口用Iceberg做統(tǒng)一的存儲(chǔ)和元數(shù)據(jù)管理。實(shí)時(shí)鏈路通過Flink流作業(yè)持續(xù)寫入Iceberg表離線鏈路通過Flink批作業(yè)或者Spark讀取同一張Iceberg表做回溯分析。兩條鏈路看到的是同一份數(shù)據(jù)、同一套schema口徑天然一致。在同程藝龍的實(shí)踐場景里我們面對(duì)的是旅游行業(yè)典型的數(shù)據(jù)特征訂單、用戶行為、庫存、價(jià)格這些核心業(yè)務(wù)數(shù)據(jù)既有實(shí)時(shí)監(jiān)控的需求又有T1甚至更長時(shí)間跨度的歷史分析需求。比如機(jī)票價(jià)格監(jiān)控需要秒級(jí)感知價(jià)格變化并觸發(fā)預(yù)警同時(shí)又要做過去一年的價(jià)格趨勢分析支撐收益管理策略。這種場景下流批一體不是錦上添花而是剛需。1.3 落地前的核心挑戰(zhàn)盤點(diǎn)理想很豐滿落地的時(shí)候問題一個(gè)接一個(gè)。我們?cè)谡戒侀_之前梳理了幾個(gè)必須解決的核心挑戰(zhàn)第一寫入性能與小文件問題。Flink流作業(yè)是持續(xù)寫入的默認(rèn)的checkpoint間隔如果設(shè)得比較短每次checkpoint都會(huì)生成一批新文件幾個(gè)小時(shí)下來Iceberg表里全是小文件元數(shù)據(jù)膨脹得厲害讀取性能急劇下降。這個(gè)問題不解決實(shí)時(shí)鏈路根本跑不穩(wěn)。第二schema演進(jìn)的兼容性。業(yè)務(wù)需求變化快加字段、改字段類型是家常便飯。Iceberg雖然支持schema演進(jìn)但Flink寫入端和下游讀取端對(duì)schema變更的感知時(shí)機(jī)、兼容策略需要仔細(xì)設(shè)計(jì)否則容易出現(xiàn)寫入端已經(jīng)加了字段、讀取端還沒更新導(dǎo)致報(bào)錯(cuò)的情況。第三Exactly-Once語義的保障。Flink的checkpoint機(jī)制和Iceberg的snapshot提交需要精確對(duì)齊。如果checkpoint成功了但I(xiàn)ceberg提交失敗或者反過來都會(huì)導(dǎo)致數(shù)據(jù)不一致。這中間涉及到兩階段提交協(xié)議的設(shè)計(jì)需要深入理解Flink的Sink API和Iceberg的Transaction API。第四查詢性能的優(yōu)化。Iceberg表的數(shù)據(jù)文件是列式存儲(chǔ)的Parquet理論上查詢性能不錯(cuò)但如果分區(qū)策略設(shè)計(jì)不合理、manifest文件過多、或者沒有做好數(shù)據(jù)聚簇查詢時(shí)掃描大量無關(guān)文件性能會(huì)大打折扣。這四個(gè)問題貫穿了我們整個(gè)實(shí)踐過程后面會(huì)逐一展開講具體的解決思路和實(shí)操細(xì)節(jié)。2. 核心機(jī)制拆解與關(guān)鍵參數(shù)設(shè)計(jì)2.1 Iceberg表格式的核心概念快速掃盲在深入實(shí)操之前有必要把Iceberg的幾個(gè)核心概念用大白話講清楚不然后面的參數(shù)調(diào)優(yōu)和問題排查會(huì)一頭霧水。Snapshot快照每次對(duì)Iceberg表的寫入操作都會(huì)生成一個(gè)新的snapshot相當(dāng)于給表拍了一張照片記錄了這一刻表里有哪些數(shù)據(jù)文件。讀取的時(shí)候指定snapshot ID就能讀到那個(gè)時(shí)間點(diǎn)的數(shù)據(jù)狀態(tài)。這個(gè)機(jī)制天然支持時(shí)間旅行查詢做數(shù)據(jù)回溯和審計(jì)非常方便。Manifest清單文件snapshot本身不直接記錄數(shù)據(jù)文件列表而是通過manifest文件來組織。一個(gè)manifest里記錄了一批數(shù)據(jù)文件的路徑、分區(qū)信息、列級(jí)統(tǒng)計(jì)信息比如min/max值。查詢的時(shí)候Iceberg先讀manifest list再根據(jù)查詢條件過濾manifest最后定位到具體的數(shù)據(jù)文件。這個(gè)層級(jí)結(jié)構(gòu)是Iceberg查詢性能的關(guān)鍵。Manifest List清單列表記錄當(dāng)前snapshot包含哪些manifest文件以及每個(gè)manifest的分區(qū)范圍等元信息。查詢引擎通過manifest list做第一層裁剪快速跳過不相關(guān)的manifest。Catalog目錄管理Iceberg表的元數(shù)據(jù)入口負(fù)責(zé)表的創(chuàng)建、刪除、元數(shù)據(jù)更新等操作。常見的Catalog實(shí)現(xiàn)有HiveCatalog、HadoopCatalog、RESTCatalog等。同程藝龍這邊用的是HiveCatalog主要是考慮到和現(xiàn)有Hive數(shù)倉的兼容性以及元數(shù)據(jù)管理的統(tǒng)一性。Partition Spec分區(qū)規(guī)范定義表的分區(qū)策略支持隱藏分區(qū)、分區(qū)演進(jìn)等高級(jí)特性。和Hive分區(qū)不同Iceberg的分區(qū)是可以演進(jìn)的改分區(qū)策略不需要重寫歷史數(shù)據(jù)。理解這幾個(gè)概念之后再看Flink寫入Iceberg的流程就清晰了Flink作業(yè)在checkpoint時(shí)把當(dāng)前批次的數(shù)據(jù)文件信息打包成一個(gè)新的snapshot通過Catalog提交到Iceberg表。下游讀取時(shí)根據(jù)最新的snapshot或者指定的歷史snapshot來讀取數(shù)據(jù)。2.2 Flink寫入Iceberg的兩種模式對(duì)比Flink集成Iceberg主要有兩種寫入模式append模式和upsert模式。這兩種模式的選擇直接決定了后續(xù)的數(shù)據(jù)處理邏輯和運(yùn)維復(fù)雜度。Append模式就是純追加寫入每條數(shù)據(jù)進(jìn)來直接寫到新的數(shù)據(jù)文件里不做任何去重或更新。這種模式實(shí)現(xiàn)簡單、寫入性能高適合日志類、事件類等只追加不更新的場景。但缺點(diǎn)是如果上游數(shù)據(jù)有重復(fù)或者需要更新Iceberg表里會(huì)出現(xiàn)重復(fù)數(shù)據(jù)需要下游做去重處理。Upsert模式支持根據(jù)主鍵更新數(shù)據(jù)。Flink作業(yè)在寫入時(shí)會(huì)根據(jù)主鍵判斷這條數(shù)據(jù)是新增還是更新然后生成對(duì)應(yīng)的數(shù)據(jù)文件和刪除文件。Iceberg通過equality delete file和position delete file來實(shí)現(xiàn)行級(jí)更新。這種模式適合訂單、用戶畫像等需要頻繁更新的場景但寫入性能會(huì)比append模式低一些因?yàn)樾枰S護(hù)刪除文件而且讀取時(shí)要做merge操作。我們當(dāng)時(shí)的選型策略是ODS層用append模式DWD和DWS層用upsert模式。ODS層接的是原始日志和binlog數(shù)據(jù)只追加不更新用append模式寫入性能最優(yōu)。DWD層做清洗和維度關(guān)聯(lián)后需要根據(jù)主鍵做更新用upsert模式保證數(shù)據(jù)唯一性。DWS層做聚合指標(biāo)也是upsert模式按維度主鍵更新指標(biāo)值。這里有個(gè)實(shí)操心得upsert模式的主鍵選擇非常關(guān)鍵。主鍵字段最好是分區(qū)字段的子集或者有強(qiáng)關(guān)聯(lián)這樣Iceberg在寫入時(shí)能更好地做文件裁剪和合并。如果主鍵和分區(qū)完全無關(guān)寫入時(shí)會(huì)產(chǎn)生大量跨分區(qū)的刪除文件性能會(huì)明顯下降。2.3 Checkpoint間隔與文件大小的平衡計(jì)算這是流式寫入Iceberg最核心的參數(shù)調(diào)優(yōu)問題。Flink的checkpoint間隔決定了多久生成一批新文件而Iceberg的數(shù)據(jù)文件大小又直接影響查詢性能。兩者需要配合調(diào)整。先看默認(rèn)情況假設(shè)checkpoint間隔是1分鐘每個(gè)checkpoint周期寫入的數(shù)據(jù)量是500MB那么每個(gè)checkpoint會(huì)生成大約5到10個(gè)Parquet文件取決于并行度。一天下來就是1440個(gè)checkpoint產(chǎn)生上萬個(gè)文件。每個(gè)文件平均50MB到100MB雖然不算特別小但manifest文件的數(shù)量會(huì)膨脹得很快查詢時(shí)讀取manifest的開銷就會(huì)變得不可忽略。我們的調(diào)優(yōu)思路是讓每個(gè)checkpoint周期寫入的數(shù)據(jù)量盡量接近一個(gè)合理的數(shù)據(jù)文件大小。Iceberg社區(qū)推薦的Parquet文件大小是128MB到512MB之間我們?nèi)≈虚g值256MB作為目標(biāo)。然后反推checkpoint間隔。具體計(jì)算過程假設(shè)Flink作業(yè)的寫入吞吐是每秒10MB那么要攢夠256MB需要大約25.6秒?？紤]到checkpoint本身有開銷以及數(shù)據(jù)寫入的波動(dòng)我們把checkpoint間隔設(shè)為30秒。這樣每個(gè)checkpoint周期大約寫入300MB數(shù)據(jù)生成1到2個(gè)Parquet文件文件大小在150MB到300MB之間比較合理。但這里有個(gè)矛盾checkpoint間隔太長故障恢復(fù)時(shí)需要重放的數(shù)據(jù)就越多恢復(fù)時(shí)間越長。30秒的間隔意味著最壞情況下要重放30秒的數(shù)據(jù)對(duì)于大部分場景是可以接受的。如果業(yè)務(wù)對(duì)恢復(fù)時(shí)間要求極高可以適當(dāng)縮短checkpoint間隔同時(shí)通過調(diào)整Flink的寫入并行度來控制每個(gè)并行子任務(wù)寫入的數(shù)據(jù)量讓每個(gè)子任務(wù)在每個(gè)checkpoint周期寫入的數(shù)據(jù)量接近目標(biāo)文件大小。還有一個(gè)參數(shù)是寫入并行度。并行度越高每個(gè)子任務(wù)分到的數(shù)據(jù)越少生成的文件越多但每個(gè)文件越小。我們的經(jīng)驗(yàn)是寫入并行度不要超過Kafka分區(qū)數(shù)否則會(huì)有子任務(wù)空閑。同時(shí)如果發(fā)現(xiàn)生成的文件普遍偏小可以適當(dāng)降低并行度讓每個(gè)子任務(wù)攢更多數(shù)據(jù)再寫。2.4 Schema演進(jìn)策略與兼容性設(shè)計(jì)Schema演進(jìn)是數(shù)據(jù)湖相比傳統(tǒng)數(shù)倉的一大優(yōu)勢但用不好也會(huì)踩坑。Iceberg支持的操作包括新增列、刪除列、重命名列、修改列類型有限制、調(diào)整列順序等。Flink寫入端和下游讀取端對(duì)schema變更的感知是異步的需要設(shè)計(jì)好兼容策略。我們的做法是所有schema變更走統(tǒng)一的元數(shù)據(jù)管理流程。新增字段時(shí)先在Iceberg表上執(zhí)行ALTER TABLE ADD COLUMN然后更新Flink寫入作業(yè)的schema定義最后通知下游讀取方更新。這個(gè)順序不能亂否則寫入端會(huì)報(bào)字段不存在的錯(cuò)誤。對(duì)于列類型修改Iceberg支持int到long、float到double這種安全的類型提升但不支持反向轉(zhuǎn)換。我們規(guī)定只允許類型提升不允許類型收窄。如果業(yè)務(wù)確實(shí)需要收窄類型走新建表加數(shù)據(jù)遷移的流程不在原表上直接改。還有一個(gè)容易忽略的點(diǎn)Flink SQL中定義的schema和Iceberg表的schema要定期做一致性校驗(yàn)。我們遇到過Flink作業(yè)的schema定義和Iceberg表實(shí)際schema不一致的情況原因是有人直接在Iceberg表上加了字段但忘了更新Flink作業(yè)。后來我們加了一個(gè)定時(shí)校驗(yàn)任務(wù)每天對(duì)比Flink作業(yè)的schema和Iceberg表的schema不一致就告警。2.5 小文件治理的自動(dòng)與手動(dòng)策略小文件是流式寫入數(shù)據(jù)湖的頑疾。Iceberg提供了幾種治理手段我們根據(jù)數(shù)據(jù)的重要程度和查詢頻率采用了分層策略。對(duì)于高優(yōu)表核心業(yè)務(wù)表查詢頻繁我們開啟了Iceberg的自動(dòng)合并功能。在Flink寫入端配置write.distribution-mode為hash讓相同分區(qū)鍵的數(shù)據(jù)盡量寫到同一個(gè)文件里。同時(shí)在Iceberg側(cè)配置定期執(zhí)行rewrite_data_files和rewrite_manifests操作把小于目標(biāo)大小的文件合并成大文件清理過期的manifest。對(duì)于普通表日志類、歸檔類我們采用手動(dòng)合并策略每天凌晨低峰期跑一次合并任務(wù)。合并任務(wù)的參數(shù)需要根據(jù)表的數(shù)據(jù)量和查詢模式來調(diào)。比如target-file-size-bytes設(shè)為256MBmin-file-size-bytes設(shè)為64MB小于64MB的文件會(huì)被合并。max-file-size-bytes設(shè)為512MB超過這個(gè)大小的文件不再合并。對(duì)于極冷表很少查詢的歷史歸檔我們不做合并只做定期的snapshot過期清理刪除不再需要的舊snapshot和對(duì)應(yīng)的數(shù)據(jù)文件釋放存儲(chǔ)空間。這里有個(gè)實(shí)操細(xì)節(jié)合并操作本身也會(huì)消耗計(jì)算資源。如果合并任務(wù)和實(shí)時(shí)寫入任務(wù)跑在同一個(gè)集群上可能會(huì)互相影響。我們的做法是把合并任務(wù)調(diào)度到獨(dú)立的批處理集群上通過讀取Iceberg表的snapshot來做合并不影響實(shí)時(shí)寫入鏈路。3. 完整實(shí)操流程與核心環(huán)節(jié)實(shí)現(xiàn)3.1 環(huán)境準(zhǔn)備與依賴版本鎖定在開始集成之前版本兼容性是第一個(gè)要確認(rèn)的事情。Flink和Iceberg的版本對(duì)應(yīng)關(guān)系比較敏感版本不匹配會(huì)導(dǎo)致各種奇怪的報(bào)錯(cuò)。我們當(dāng)時(shí)用的組合是Flink 1.13.2 Iceberg 0.12.0 Hadoop 3.2.1。這個(gè)組合是經(jīng)過社區(qū)驗(yàn)證的穩(wěn)定版本。依賴方面需要在Flink的lib目錄下放入以下jar包# Iceberg核心包 iceberg-flink-runtime-0.12.0.jar # Hive Catalog依賴如果使用HiveCatalog iceberg-hive-runtime-0.12.0.jar # Hadoop依賴如果Flink集群沒有自帶 hadoop-common-3.2.1.jar hadoop-hdfs-client-3.2.1.jar注意iceberg-flink-runtime是一個(gè)shaded包已經(jīng)包含了大部分依賴不要重復(fù)引入Iceberg的其他模塊否則會(huì)出現(xiàn)類沖突。Hive Metastore的配置需要在Flink的sql-client-defaults.yaml或者代碼中指定catalogs: - name: iceberg_catalog type: iceberg catalog-type: hive uri: thrift://your-hive-metastore:9083 warehouse: hdfs://your-namenode:8020/warehouse/iceberg3.2 Flink SQL建表與參數(shù)配置詳解建表是第一步也是參數(shù)配置最集中的地方。下面是一個(gè)典型的upsert模式Iceberg表的建表語句CREATE TABLE iceberg_catalog.db.order_detail ( order_id BIGINT, user_id BIGINT, product_id BIGINT, order_amount DECIMAL(10,2), order_status STRING, create_time TIMESTAMP(3), update_time TIMESTAMP(3), dt STRING, PRIMARY KEY (order_id) NOT ENFORCED ) PARTITIONED BY (dt) WITH ( connector iceberg, catalog-type hive, catalog-database db, catalog-table order_detail, write.distribution-mode hash, write.upsert.enabled true, write.metadata.delete-after-commit.enabled true, write.metadata.previous-versions-max 10, write.format.default parquet, write.parquet.compression-codec zstd, write.target-file-size-bytes 268435456 );幾個(gè)關(guān)鍵參數(shù)的解釋write.distribution-mode設(shè)為hash表示按照分區(qū)鍵做hash分發(fā)相同分區(qū)的數(shù)據(jù)會(huì)寫到同一個(gè)Flink子任務(wù)減少小文件數(shù)量。如果設(shè)為none數(shù)據(jù)隨機(jī)分發(fā)小文件會(huì)更多。write.upsert.enabled設(shè)為true開啟upsert模式。注意這個(gè)參數(shù)在Flink SQL中是通過主鍵定義隱式開啟的如果定義了PRIMARY KEYIceberg會(huì)自動(dòng)使用upsert模式。write.metadata.delete-after-commit.enabled設(shè)為true每次提交后自動(dòng)刪除過期的元數(shù)據(jù)文件防止元數(shù)據(jù)無限膨脹。write.metadata.previous-versions-max設(shè)為10保留最近10個(gè)版本的元數(shù)據(jù)支持時(shí)間旅行查詢最近10個(gè)snapshot。write.target-file-size-bytes設(shè)為256MB這是目標(biāo)文件大小Iceberg在寫入時(shí)會(huì)盡量讓每個(gè)文件接近這個(gè)大小。write.parquet.compression-codec設(shè)為zstd壓縮比和性能比較均衡。如果對(duì)存儲(chǔ)成本敏感可以用zstd如果對(duì)讀取性能要求極高可以用snappy。3.3 實(shí)時(shí)寫入作業(yè)的提交與調(diào)優(yōu)建完表之后提交Flink流作業(yè)寫入數(shù)據(jù)。作業(yè)的并行度設(shè)置需要根據(jù)數(shù)據(jù)量和集群資源來定。我們的經(jīng)驗(yàn)值是每個(gè)并行子任務(wù)的寫入吞吐控制在5MB/s到10MB/s之間。如果總吞吐是100MB/s那么并行度設(shè)為10到20比較合適。Checkpoint的配置在Flink作業(yè)的配置文件中execution.checkpointing.interval: 30s execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 10s execution.checkpointing.max-concurrent-checkpoints: 1 state.backend: rocksdb state.backend.incremental: trueexecution.checkpointing.interval設(shè)為30秒和前面計(jì)算的文件大小目標(biāo)對(duì)應(yīng)。execution.checkpointing.timeout設(shè)為10分鐘給Iceberg提交snapshot留足時(shí)間。如果超時(shí)時(shí)間設(shè)得太短Iceberg提交還沒完成checkpoint就超時(shí)了會(huì)導(dǎo)致作業(yè)頻繁重啟。execution.checkpointing.max-concurrent-checkpoints設(shè)為1禁止并發(fā)checkpoint。Iceberg的snapshot提交需要串行執(zhí)行并發(fā)checkpoint會(huì)導(dǎo)致提交沖突。state.backend用RocksDB開啟增量checkpoint減少狀態(tài)存儲(chǔ)的開銷。3.4 離線讀取與時(shí)間旅行查詢實(shí)操Iceberg表建好、數(shù)據(jù)寫進(jìn)去之后離線讀取就很簡單了。用Flink SQL或者Spark SQL都可以直接查-- Flink SQL查詢最新數(shù)據(jù) SELECT * FROM iceberg_catalog.db.order_detail WHERE dt 2024-01-15; -- Flink SQL時(shí)間旅行查詢查詢某個(gè)snapshot的數(shù)據(jù) SELECT * FROM iceberg_catalog.db.order_detail /* OPTIONS(snapshot-id1234567890) */ WHERE dt 2024-01-15;時(shí)間旅行查詢?cè)谂挪閿?shù)據(jù)問題時(shí)特別有用。比如業(yè)務(wù)方反饋某天的數(shù)據(jù)有問題我們可以通過時(shí)間旅行查詢到那天的snapshot對(duì)比前后幾個(gè)snapshot的數(shù)據(jù)變化快速定位是哪次寫入引入了異常數(shù)據(jù)。還有一個(gè)實(shí)用技巧通過Iceberg的元數(shù)據(jù)表查詢表的文件分布情況。Iceberg提供了snapshots、manifests、files等元數(shù)據(jù)表可以直接用SQL查詢-- 查詢表的所有snapshot SELECT * FROM iceberg_catalog.db.order_detail.snapshots; -- 查詢某個(gè)snapshot的文件列表 SELECT * FROM iceberg_catalog.db.order_detail.files WHERE snapshot_id 1234567890;這些元數(shù)據(jù)表在排查小文件問題、分析數(shù)據(jù)分布時(shí)非常有用。3.5 監(jiān)控指標(biāo)與告警配置生產(chǎn)環(huán)境跑起來之后監(jiān)控是必不可少的。我們主要監(jiān)控以下幾類指標(biāo)Flink作業(yè)層面checkpoint成功率、checkpoint耗時(shí)、背壓情況、吞吐量。這些指標(biāo)通過Flink的Metric系統(tǒng)采集接入Prometheus和Grafana做可視化。Iceberg表層面snapshot數(shù)量、manifest文件數(shù)量、數(shù)據(jù)文件數(shù)量、平均文件大小、元數(shù)據(jù)文件大小。這些指標(biāo)通過定時(shí)查詢Iceberg的元數(shù)據(jù)表來采集。存儲(chǔ)層面HDFS的存儲(chǔ)使用量、文件數(shù)量增長趨勢。如果發(fā)現(xiàn)文件數(shù)量增長過快說明小文件問題嚴(yán)重需要觸發(fā)合并任務(wù)。告警規(guī)則方面我們?cè)O(shè)置了幾個(gè)關(guān)鍵閾值checkpoint連續(xù)失敗3次告警、單次checkpoint耗時(shí)超過5分鐘告警、Iceberg表的小文件數(shù)量小于64MB的文件超過1000個(gè)告警、元數(shù)據(jù)文件大小超過1GB告警。4. 常見問題與排查技巧實(shí)錄4.1 寫入性能突然下降的排查路徑這是我們?cè)谏a(chǎn)環(huán)境遇到最多的問題。表現(xiàn)是Flink作業(yè)的吞吐量突然下降checkpoint耗時(shí)變長背壓升高。排查路徑如下第一步看Flink的背壓情況。通過Flink Web UI或者M(jìn)etric系統(tǒng)查看各個(gè)算子的背壓狀態(tài)。如果背壓出現(xiàn)在Iceberg Sink算子說明寫入Iceberg這一側(cè)有問題。第二步看Iceberg表的文件數(shù)量。如果文件數(shù)量在短時(shí)間內(nèi)快速增長說明小文件問題嚴(yán)重了。大量小文件會(huì)導(dǎo)致Iceberg在提交snapshot時(shí)需要處理大量的manifest文件提交耗時(shí)變長進(jìn)而拖慢整個(gè)checkpoint。第三步看HDFS的寫入延遲。如果HDFS的寫入延遲升高也會(huì)導(dǎo)致Iceberg寫入變慢。這種情況通常是HDFS集群本身的問題比如NameNode負(fù)載過高、DataNode磁盤IO瓶頸等。第四步看Iceberg的提交日志。Iceberg在提交snapshot時(shí)會(huì)打印詳細(xì)的日志包括生成了多少個(gè)數(shù)據(jù)文件、多少個(gè)刪除文件、提交耗時(shí)等。通過日志可以定位到具體的瓶頸。我們遇到過一次典型的性能下降原因是某個(gè)Flink作業(yè)的并行度從10調(diào)到了20導(dǎo)致每個(gè)子任務(wù)寫入的數(shù)據(jù)量減半生成的文件數(shù)量翻倍小文件問題加劇最終拖慢了整個(gè)寫入鏈路。解決辦法是把并行度調(diào)回10同時(shí)開啟自動(dòng)合并。4.2 Checkpoint失敗與數(shù)據(jù)不一致的修復(fù)Checkpoint失敗是流作業(yè)的常見問題但和Iceberg集成后失敗的原因和修復(fù)方式有一些特殊性。場景一Iceberg提交超時(shí)導(dǎo)致checkpoint失敗。表現(xiàn)是checkpoint超時(shí)日志里能看到Iceberg的commit操作還在執(zhí)行。這種情況通常是Iceberg提交的數(shù)據(jù)量太大或者HDFS寫入太慢。解決辦法是縮短checkpoint間隔減少每次提交的數(shù)據(jù)量或者增加checkpoint的超時(shí)時(shí)間。場景二Iceberg提交成功但checkpoint失敗。這種情況比較棘手因?yàn)镮ceberg已經(jīng)提交了snapshot但Flink認(rèn)為checkpoint失敗了會(huì)回滾到上一個(gè)checkpoint。這會(huì)導(dǎo)致Iceberg表里多了一批“孤兒數(shù)據(jù)”雖然不影響讀取因?yàn)樽x取是基于snapshot的但會(huì)浪費(fèi)存儲(chǔ)空間。解決辦法是配置Iceberg的write.metadata.delete-after-commit.enabled為true定期清理孤兒snapshot。場景三Checkpoint成功但I(xiàn)ceberg提交失敗。這種情況會(huì)導(dǎo)致數(shù)據(jù)丟失因?yàn)镕link認(rèn)為checkpoint成功了不會(huì)重放這部分?jǐn)?shù)據(jù)但I(xiàn)ceberg里沒有寫入。這種情況通常是由于Iceberg的Catalog不可用或者權(quán)限問題導(dǎo)致的。解決辦法是配置重試機(jī)制Iceberg的commit操作支持重試可以在Flink的Sink配置中設(shè)置重試次數(shù)和重試間隔。實(shí)操心得我們后來在Flink作業(yè)里加了一個(gè)“提交確認(rèn)”機(jī)制每次checkpoint完成后額外做一次Iceberg的snapshot校驗(yàn)確認(rèn)最新的snapshot ID和checkpoint的ID對(duì)應(yīng)。如果不一致觸發(fā)告警并暫停作業(yè)人工介入排查。4.3 小文件過多的緊急處理方案小文件問題如果已經(jīng)積累到影響查詢性能的程度需要緊急處理。我們的處理流程是第一步評(píng)估影響范圍。查詢Iceberg的files元數(shù)據(jù)表統(tǒng)計(jì)小于64MB的文件數(shù)量和總大小評(píng)估合并需要的時(shí)間和資源。第二步暫停實(shí)時(shí)寫入。如果小文件問題非常嚴(yán)重合并任務(wù)和實(shí)時(shí)寫入同時(shí)跑會(huì)互相影響建議先暫停實(shí)時(shí)寫入作業(yè)等合并完成后再恢復(fù)。暫停時(shí)間根據(jù)合并的數(shù)據(jù)量來定一般幾十分鐘到幾個(gè)小時(shí)。第三步執(zhí)行合并任務(wù)。用Spark或者Flink批作業(yè)執(zhí)行Iceberg的rewrite_data_files操作-- Spark SQL執(zhí)行合并 CALL iceberg_catalog.system.rewrite_data_files( table db.order_detail, options map( target-file-size-bytes, 268435456, min-file-size-bytes, 67108864, max-file-size-bytes, 536870912 ) );第四步清理過期元數(shù)據(jù)。合并完成后執(zhí)行rewrite_manifests和expire_snapshots清理不再需要的manifest文件和snapshot。第五步恢復(fù)實(shí)時(shí)寫入。確認(rèn)合并完成后恢復(fù)Flink作業(yè)并調(diào)整參數(shù)防止小文件問題再次出現(xiàn)。4.4 Schema變更導(dǎo)致作業(yè)失敗的應(yīng)急處理Schema變更導(dǎo)致的作業(yè)失敗通常發(fā)生在寫入端和讀取端的schema不一致時(shí)。應(yīng)急處理流程情況一寫入端報(bào)字段不存在。原因是Iceberg表加了字段但Flink作業(yè)的schema沒更新。解決辦法是更新Flink作業(yè)的schema定義并重啟作業(yè)。如果作業(yè)不能立即重啟可以先用ALTER TABLE DROP COLUMN把新加的字段刪掉等作業(yè)更新后再加回來。情況二讀取端報(bào)字段類型不匹配。原因是Iceberg表的字段類型改了但讀取端的schema沒更新。解決辦法是更新讀取端的schema定義。如果讀取端是BI工具或者即席查詢需要通知相關(guān)方更新。情況三寫入端報(bào)主鍵沖突。原因是upsert模式的主鍵定義和Iceberg表的主鍵定義不一致。解決辦法是檢查兩邊的定義確保完全一致。避坑技巧我們后來規(guī)定所有schema變更必須走工單流程變更前先在測試環(huán)境驗(yàn)證確認(rèn)寫入端和讀取端都能正常工作后再上生產(chǎn)。同時(shí)在Flink作業(yè)里加了schema兼容性檢查作業(yè)啟動(dòng)時(shí)對(duì)比Flink schema和Iceberg schema不一致直接拒絕啟動(dòng)。4.5 常見問題速查表問題現(xiàn)象可能原因排查方法解決方案寫入吞吐突然下降小文件過多、HDFS延遲高查看Iceberg文件數(shù)量、HDFS監(jiān)控執(zhí)行合并、調(diào)整并行度Checkpoint頻繁超時(shí)單次提交數(shù)據(jù)量太大查看checkpoint耗時(shí)、Iceberg提交日志縮短checkpoint間隔、增加超時(shí)時(shí)間數(shù)據(jù)重復(fù)upsert模式未生效檢查主鍵定義、查詢Iceberg表數(shù)據(jù)確認(rèn)主鍵定義、開啟upsert查詢性能差分區(qū)裁剪失效、文件過大查看查詢計(jì)劃、文件大小分布調(diào)整分區(qū)策略、合并大文件元數(shù)據(jù)膨脹snapshot未清理查看snapshots元數(shù)據(jù)表配置自動(dòng)清理、執(zhí)行expire_snapshots作業(yè)啟動(dòng)失敗schema不一致、依賴沖突查看啟動(dòng)日志、對(duì)比schema更新schema、檢查jar包版本時(shí)間旅行查詢報(bào)錯(cuò)snapshot已過期查看snapshots元數(shù)據(jù)表調(diào)整snapshot保留策略5. 生產(chǎn)環(huán)境運(yùn)維經(jīng)驗(yàn)與性能優(yōu)化心得5.1 資源規(guī)劃與容量評(píng)估方法Iceberg表的存儲(chǔ)容量增長是可以預(yù)估的。核心公式是日增數(shù)據(jù)量 × 副本數(shù) × 壓縮比 × 保留天數(shù)。比如日增原始數(shù)據(jù)100GBHDFS副本數(shù)3Parquet壓縮比0.3保留365天那么總存儲(chǔ)需求大約是100GB × 3 × 0.3 × 365 ≈ 32TB。這還不包括合并操作產(chǎn)生的臨時(shí)文件和舊snapshot占用的空間實(shí)際規(guī)劃時(shí)要留出30%到50%的余量。計(jì)算資源方面Flink寫入作業(yè)的資源需求主要取決于寫入吞吐和checkpoint的數(shù)據(jù)量。我們的經(jīng)驗(yàn)值是每10MB/s的寫入吞吐需要大約4個(gè)CPU核心和8GB內(nèi)存。這個(gè)值會(huì)根據(jù)數(shù)據(jù)復(fù)雜度、序列化開銷、Iceberg提交開銷等因素浮動(dòng)建議在實(shí)際壓測后確定。合并任務(wù)的資源需求取決于表的數(shù)據(jù)量和合并頻率。每天合并一次的話合并任務(wù)的處理能力建議是日增數(shù)據(jù)量的2到3倍確保能在低峰期內(nèi)完成合并。5.2 多表共用一個(gè)Flink作業(yè)的利弊分析為了節(jié)省資源我們一開始嘗試把多個(gè)Iceberg表的寫入放在同一個(gè)Flink作業(yè)里。好處是節(jié)省了TaskManager的資源減少了作業(yè)數(shù)量運(yùn)維成本低。但壞處也很明顯故障隔離性差。一張表的寫入出現(xiàn)問題整個(gè)作業(yè)都會(huì)受影響所有表的數(shù)據(jù)寫入都會(huì)延遲。資源競爭。不同表的寫入吞吐不同有的表數(shù)據(jù)量大、有的表數(shù)據(jù)量小共用一個(gè)作業(yè)會(huì)導(dǎo)致資源分配不均數(shù)據(jù)量大的表可能搶不到足夠的資源。Checkpoint相互影響。一個(gè)作業(yè)的checkpoint是全局的如果某張表的Iceberg提交特別慢會(huì)拖慢整個(gè)checkpoint影響所有表。后來我們改成了按業(yè)務(wù)域拆分作業(yè)每個(gè)業(yè)務(wù)域的核心表放在一個(gè)作業(yè)里非核心表單獨(dú)跑。這樣既保證了核心表的穩(wěn)定性又控制了作業(yè)數(shù)量。5.3 數(shù)據(jù)回溯與重跑的正確姿勢數(shù)據(jù)回溯是數(shù)據(jù)平臺(tái)的剛需Iceberg在這方面提供了很好的支持。我們的回溯流程是第一步確定回溯范圍。通過時(shí)間旅行查詢確定要回溯到哪個(gè)snapshot以及需要重跑的時(shí)間段。第二步暫停下游消費(fèi)。通知下游停止消費(fèi)Iceberg表的數(shù)據(jù)避免讀到中間狀態(tài)的數(shù)據(jù)。第三步執(zhí)行回溯寫入。用Flink批作業(yè)或者Spark作業(yè)重新計(jì)算指定時(shí)間段的數(shù)據(jù)寫入Iceberg表。注意寫入時(shí)要指定新的分區(qū)或者使用INSERT OVERWRITE語義避免和現(xiàn)有數(shù)據(jù)沖突。第四步校驗(yàn)數(shù)據(jù)?；厮萃瓿珊髮?duì)比回溯前后的數(shù)據(jù)量和關(guān)鍵指標(biāo)確認(rèn)數(shù)據(jù)正確。第五步恢復(fù)下游消費(fèi)。通知下游恢復(fù)消費(fèi)并告知回溯完成的時(shí)間點(diǎn)。實(shí)操心得回溯操作最好在低峰期執(zhí)行并且提前通知相關(guān)方。我們有一次在業(yè)務(wù)高峰期做回溯導(dǎo)致Iceberg表的寫入壓力驟增影響了實(shí)時(shí)鏈路的性能。后來規(guī)定所有回溯操作必須提前一天報(bào)備安排在凌晨執(zhí)行。5.4 從Iceberg 0.12到0.14的升級(jí)踩坑記錄Iceberg社區(qū)迭代很快我們后來從0.12升級(jí)到了0.14中間踩了一些坑記錄一下坑一Flink版本兼容性。Iceberg 0.14要求Flink 1.14及以上版本我們當(dāng)時(shí)用的Flink 1.13.2不兼容需要先升級(jí)Flink。升級(jí)Flink又涉及到狀態(tài)兼容性問題需要做狀態(tài)遷移?？佣嗀PI變更。Iceberg 0.14對(duì)Flink SQL的Connector參數(shù)做了一些調(diào)整比如write.upsert.enabled參數(shù)被廢棄改為通過主鍵定義自動(dòng)識(shí)別。升級(jí)后需要修改建表語句?？尤獢?shù)據(jù)格式變更。Iceberg 0.14對(duì)元數(shù)據(jù)格式做了一些優(yōu)化升級(jí)后需要執(zhí)行一次元數(shù)據(jù)遷移操作。這個(gè)操作是不可逆的升級(jí)前一定要備份元數(shù)據(jù)。坑四性能表現(xiàn)差異。升級(jí)后發(fā)現(xiàn)寫入性能比0.12略有下降排查后發(fā)現(xiàn)是0.14默認(rèn)開啟了更嚴(yán)格的元數(shù)據(jù)校驗(yàn)。通過調(diào)整write.metadata.compression-codec參數(shù)把元數(shù)據(jù)壓縮算法從gzip改為zstd性能恢復(fù)到了升級(jí)前的水平。5.5 未來演進(jìn)方向與擴(kuò)展思路Iceberg和Flink的集成還在快速演進(jìn)中我們也在持續(xù)關(guān)注幾個(gè)方向一是Flink CDC與Iceberg的深度集成。目前我們是從Kafka接binlog數(shù)據(jù)寫入Iceberg鏈路比較長。Flink CDC可以直接從數(shù)據(jù)庫抓取變更數(shù)據(jù)寫入Iceberg鏈路更短、延遲更低。我們正在測試這個(gè)方案初步效果不錯(cuò)。二是Iceberg的Row-level Delete優(yōu)化。Iceberg 0.14引入了更高效的刪除文件格式upsert模式的寫入性能有明顯提升。我們計(jì)劃在下一個(gè)版本升級(jí)中驗(yàn)證這個(gè)特性。三是多引擎共享Iceberg表。目前我們主要是Flink寫入、Flink和Spark讀取。未來計(jì)劃讓Trino、Presto等引擎也能直接查詢Iceberg表實(shí)現(xiàn)真正的多引擎共享。四是冷熱數(shù)據(jù)分層存儲(chǔ)。Iceberg支持自定義文件存儲(chǔ)路徑可以把冷數(shù)據(jù)放到更便宜的存儲(chǔ)介質(zhì)上熱數(shù)據(jù)保留在高性能存儲(chǔ)上。這個(gè)特性對(duì)降低存儲(chǔ)成本很有幫助我們正在做方案設(shè)計(jì)。整體來看Flink加Iceberg的組合在同程藝龍的實(shí)踐是成功的。流批一體不是一句口號(hào)而是實(shí)實(shí)在在解決了我們數(shù)據(jù)口徑不一致、鏈路重復(fù)建設(shè)的問題。當(dāng)然落地過程中遇到的坑也不少上面這些經(jīng)驗(yàn)都是一個(gè)個(gè)問題踩出來的。如果你也在做類似的實(shí)踐希望這些內(nèi)容能幫你少走一些彎路。 Flink+Iceberg流批一體數(shù)據(jù)湖實(shí)戰(zhàn)：選型、調(diào)優(yōu)與運(yùn)維
1. 流批一體場景下的數(shù)據(jù)湖選型思考1.1 從Lambda架構(gòu)到流批一體的演進(jìn)邏輯做數(shù)據(jù)平臺(tái)的人都有一個(gè)共同的痛同一份業(yè)務(wù)邏輯離線鏈路寫一遍實(shí)時(shí)鏈路再寫一遍兩套代碼、兩套運(yùn)維、兩套口徑。早期大家用Lambda架構(gòu)扛著離線跑T1出報(bào)表實(shí)時(shí)鏈路單獨(dú)維護(hù)一套指標(biāo)時(shí)間一長兩條鏈路的數(shù)據(jù)對(duì)不上業(yè)務(wù)方找過來排查成本極高。后來大家開始琢磨Kappa架構(gòu)把所有計(jì)算都收斂到流上但流處理引擎對(duì)大規(guī)模歷史數(shù)據(jù)的回溯分析、批量修正能力又不夠成熟于是“流批一體”這個(gè)概念被反復(fù)提起。流批一體的核心訴求其實(shí)很樸素一套SQL、一份元數(shù)據(jù)、一份存儲(chǔ)既能處理無界的實(shí)時(shí)流也能處理有界的歷史批。要做到這一點(diǎn)計(jì)算引擎和存儲(chǔ)層必須解耦存儲(chǔ)層要能同時(shí)承接高頻小批量的流式寫入和大吞吐的批量讀取。這就是數(shù)據(jù)湖技術(shù)在這個(gè)階段被推到臺(tái)前的原因。在眾多數(shù)據(jù)湖方案里Iceberg、Hudi、Delta Lake是繞不開的三個(gè)選項(xiàng)。我們當(dāng)時(shí)做選型的時(shí)候核心考量維度有幾個(gè)schema演進(jìn)能力、分區(qū)裁剪效率、小文件治理機(jī)制、與計(jì)算引擎的耦合程度、以及社區(qū)活躍度。Hudi的upsert能力很強(qiáng)適合高頻更新的場景但它和Spark的綁定相對(duì)深一些Delta Lake在Databricks生態(tài)里體驗(yàn)很好但開源版本在Flink集成上的成熟度當(dāng)時(shí)還不夠。Iceberg的設(shè)計(jì)哲學(xué)是“表格式規(guī)范優(yōu)先”它把表的元數(shù)據(jù)管理抽象得非常干凈計(jì)算引擎只需要按照規(guī)范去讀寫就行這種解耦特性讓它天然適合做流批一體的存儲(chǔ)底座。1.2 為什么是Flink加Iceberg的組合Flink在流處理上的優(yōu)勢不用多說真正的流式語義、事件時(shí)間處理、精確一次的狀態(tài)一致性這些都是它的看家本領(lǐng)。但Flink早期在批處理上的表現(xiàn)只能算“能用”直到1.12之后批流一體的SQL層逐漸成熟Flink才真正具備了統(tǒng)一計(jì)算引擎的底氣。Iceberg這邊它的核心抽象是表格式Table Format不是存儲(chǔ)引擎也不是文件格式。它定義了一套元數(shù)據(jù)規(guī)范包括snapshot、manifest、manifest list這些概念把“哪些文件屬于這張表”“這些文件在某個(gè)時(shí)間點(diǎn)的狀態(tài)是什么”管理得清清楚楚。這種設(shè)計(jì)帶來的好處是寫入方和讀取方完全解耦寫入方只管生成新的snapshot讀取方根據(jù)snapshot做快照隔離讀取互不干擾。Flink加Iceberg的組合本質(zhì)上是用Flink做統(tǒng)一的寫入和計(jì)算入口用Iceberg做統(tǒng)一的存儲(chǔ)和元數(shù)據(jù)管理。實(shí)時(shí)鏈路通過Flink流作業(yè)持續(xù)寫入Iceberg表離線鏈路通過Flink批作業(yè)或者Spark讀取同一張Iceberg表做回溯分析。兩條鏈路看到的是同一份數(shù)據(jù)、同一套schema口徑天然一致。在同程藝龍的實(shí)踐場景里我們面對(duì)的是旅游行業(yè)典型的數(shù)據(jù)特征訂單、用戶行為、庫存、價(jià)格這些核心業(yè)務(wù)數(shù)據(jù)既有實(shí)時(shí)監(jiān)控的需求又有T1甚至更長時(shí)間跨度的歷史分析需求。比如機(jī)票價(jià)格監(jiān)控需要秒級(jí)感知價(jià)格變化并觸發(fā)預(yù)警同時(shí)又要做過去一年的價(jià)格趨勢分析支撐收益管理策略。這種場景下流批一體不是錦上添花而是剛需。1.3 落地前的核心挑戰(zhàn)盤點(diǎn)理想很豐滿落地的時(shí)候問題一個(gè)接一個(gè)。我們?cè)谡戒侀_之前梳理了幾個(gè)必須解決的核心挑戰(zhàn)第一寫入性能與小文件問題。Flink流作業(yè)是持續(xù)寫入的默認(rèn)的checkpoint間隔如果設(shè)得比較短每次checkpoint都會(huì)生成一批新文件幾個(gè)小時(shí)下來Iceberg表里全是小文件元數(shù)據(jù)膨脹得厲害讀取性能急劇下降。這個(gè)問題不解決實(shí)時(shí)鏈路根本跑不穩(wěn)。第二schema演進(jìn)的兼容性。業(yè)務(wù)需求變化快加字段、改字段類型是家常便飯。Iceberg雖然支持schema演進(jìn)但Flink寫入端和下游讀取端對(duì)schema變更的感知時(shí)機(jī)、兼容策略需要仔細(xì)設(shè)計(jì)否則容易出現(xiàn)寫入端已經(jīng)加了字段、讀取端還沒更新導(dǎo)致報(bào)錯(cuò)的情況。第三Exactly-Once語義的保障。Flink的checkpoint機(jī)制和Iceberg的snapshot提交需要精確對(duì)齊。如果checkpoint成功了但I(xiàn)ceberg提交失敗或者反過來都會(huì)導(dǎo)致數(shù)據(jù)不一致。這中間涉及到兩階段提交協(xié)議的設(shè)計(jì)需要深入理解Flink的Sink API和Iceberg的Transaction API。第四查詢性能的優(yōu)化。Iceberg表的數(shù)據(jù)文件是列式存儲(chǔ)的Parquet理論上查詢性能不錯(cuò)但如果分區(qū)策略設(shè)計(jì)不合理、manifest文件過多、或者沒有做好數(shù)據(jù)聚簇查詢時(shí)掃描大量無關(guān)文件性能會(huì)大打折扣。這四個(gè)問題貫穿了我們整個(gè)實(shí)踐過程后面會(huì)逐一展開講具體的解決思路和實(shí)操細(xì)節(jié)。2. 核心機(jī)制拆解與關(guān)鍵參數(shù)設(shè)計(jì)2.1 Iceberg表格式的核心概念快速掃盲在深入實(shí)操之前有必要把Iceberg的幾個(gè)核心概念用大白話講清楚不然后面的參數(shù)調(diào)優(yōu)和問題排查會(huì)一頭霧水。Snapshot快照每次對(duì)Iceberg表的寫入操作都會(huì)生成一個(gè)新的snapshot相當(dāng)于給表拍了一張照片記錄了這一刻表里有哪些數(shù)據(jù)文件。讀取的時(shí)候指定snapshot ID就能讀到那個(gè)時(shí)間點(diǎn)的數(shù)據(jù)狀態(tài)。這個(gè)機(jī)制天然支持時(shí)間旅行查詢做數(shù)據(jù)回溯和審計(jì)非常方便。Manifest清單文件snapshot本身不直接記錄數(shù)據(jù)文件列表而是通過manifest文件來組織。一個(gè)manifest里記錄了一批數(shù)據(jù)文件的路徑、分區(qū)信息、列級(jí)統(tǒng)計(jì)信息比如min/max值。查詢的時(shí)候Iceberg先讀manifest list再根據(jù)查詢條件過濾manifest最后定位到具體的數(shù)據(jù)文件。這個(gè)層級(jí)結(jié)構(gòu)是Iceberg查詢性能的關(guān)鍵。Manifest List清單列表記錄當(dāng)前snapshot包含哪些manifest文件以及每個(gè)manifest的分區(qū)范圍等元信息。查詢引擎通過manifest list做第一層裁剪快速跳過不相關(guān)的manifest。Catalog目錄管理Iceberg表的元數(shù)據(jù)入口負(fù)責(zé)表的創(chuàng)建、刪除、元數(shù)據(jù)更新等操作。常見的Catalog實(shí)現(xiàn)有HiveCatalog、HadoopCatalog、RESTCatalog等。同程藝龍這邊用的是HiveCatalog主要是考慮到和現(xiàn)有Hive數(shù)倉的兼容性以及元數(shù)據(jù)管理的統(tǒng)一性。Partition Spec分區(qū)規(guī)范定義表的分區(qū)策略支持隱藏分區(qū)、分區(qū)演進(jìn)等高級(jí)特性。和Hive分區(qū)不同Iceberg的分區(qū)是可以演進(jìn)的改分區(qū)策略不需要重寫歷史數(shù)據(jù)。理解這幾個(gè)概念之后再看Flink寫入Iceberg的流程就清晰了Flink作業(yè)在checkpoint時(shí)把當(dāng)前批次的數(shù)據(jù)文件信息打包成一個(gè)新的snapshot通過Catalog提交到Iceberg表。下游讀取時(shí)根據(jù)最新的snapshot或者指定的歷史snapshot來讀取數(shù)據(jù)。2.2 Flink寫入Iceberg的兩種模式對(duì)比Flink集成Iceberg主要有兩種寫入模式append模式和upsert模式。這兩種模式的選擇直接決定了后續(xù)的數(shù)據(jù)處理邏輯和運(yùn)維復(fù)雜度。Append模式就是純追加寫入每條數(shù)據(jù)進(jìn)來直接寫到新的數(shù)據(jù)文件里不做任何去重或更新。這種模式實(shí)現(xiàn)簡單、寫入性能高適合日志類、事件類等只追加不更新的場景。但缺點(diǎn)是如果上游數(shù)據(jù)有重復(fù)或者需要更新Iceberg表里會(huì)出現(xiàn)重復(fù)數(shù)據(jù)需要下游做去重處理。Upsert模式支持根據(jù)主鍵更新數(shù)據(jù)。Flink作業(yè)在寫入時(shí)會(huì)根據(jù)主鍵判斷這條數(shù)據(jù)是新增還是更新然后生成對(duì)應(yīng)的數(shù)據(jù)文件和刪除文件。Iceberg通過equality delete file和position delete file來實(shí)現(xiàn)行級(jí)更新。這種模式適合訂單、用戶畫像等需要頻繁更新的場景但寫入性能會(huì)比append模式低一些因?yàn)樾枰S護(hù)刪除文件而且讀取時(shí)要做merge操作。我們當(dāng)時(shí)的選型策略是ODS層用append模式DWD和DWS層用upsert模式。ODS層接的是原始日志和binlog數(shù)據(jù)只追加不更新用append模式寫入性能最優(yōu)。DWD層做清洗和維度關(guān)聯(lián)后需要根據(jù)主鍵做更新用upsert模式保證數(shù)據(jù)唯一性。DWS層做聚合指標(biāo)也是upsert模式按維度主鍵更新指標(biāo)值。這里有個(gè)實(shí)操心得upsert模式的主鍵選擇非常關(guān)鍵。主鍵字段最好是分區(qū)字段的子集或者有強(qiáng)關(guān)聯(lián)這樣Iceberg在寫入時(shí)能更好地做文件裁剪和合并。如果主鍵和分區(qū)完全無關(guān)寫入時(shí)會(huì)產(chǎn)生大量跨分區(qū)的刪除文件性能會(huì)明顯下降。2.3 Checkpoint間隔與文件大小的平衡計(jì)算這是流式寫入Iceberg最核心的參數(shù)調(diào)優(yōu)問題。Flink的checkpoint間隔決定了多久生成一批新文件而Iceberg的數(shù)據(jù)文件大小又直接影響查詢性能。兩者需要配合調(diào)整。先看默認(rèn)情況假設(shè)checkpoint間隔是1分鐘每個(gè)checkpoint周期寫入的數(shù)據(jù)量是500MB那么每個(gè)checkpoint會(huì)生成大約5到10個(gè)Parquet文件取決于并行度。一天下來就是1440個(gè)checkpoint產(chǎn)生上萬個(gè)文件。每個(gè)文件平均50MB到100MB雖然不算特別小但manifest文件的數(shù)量會(huì)膨脹得很快查詢時(shí)讀取manifest的開銷就會(huì)變得不可忽略。我們的調(diào)優(yōu)思路是讓每個(gè)checkpoint周期寫入的數(shù)據(jù)量盡量接近一個(gè)合理的數(shù)據(jù)文件大小。Iceberg社區(qū)推薦的Parquet文件大小是128MB到512MB之間我們?nèi)≈虚g值256MB作為目標(biāo)。然后反推checkpoint間隔。具體計(jì)算過程假設(shè)Flink作業(yè)的寫入吞吐是每秒10MB那么要攢夠256MB需要大約25.6秒?？紤]到checkpoint本身有開銷以及數(shù)據(jù)寫入的波動(dòng)我們把checkpoint間隔設(shè)為30秒。這樣每個(gè)checkpoint周期大約寫入300MB數(shù)據(jù)生成1到2個(gè)Parquet文件文件大小在150MB到300MB之間比較合理。但這里有個(gè)矛盾checkpoint間隔太長故障恢復(fù)時(shí)需要重放的數(shù)據(jù)就越多恢復(fù)時(shí)間越長。30秒的間隔意味著最壞情況下要重放30秒的數(shù)據(jù)對(duì)于大部分場景是可以接受的。如果業(yè)務(wù)對(duì)恢復(fù)時(shí)間要求極高可以適當(dāng)縮短checkpoint間隔同時(shí)通過調(diào)整Flink的寫入并行度來控制每個(gè)并行子任務(wù)寫入的數(shù)據(jù)量讓每個(gè)子任務(wù)在每個(gè)checkpoint周期寫入的數(shù)據(jù)量接近目標(biāo)文件大小。還有一個(gè)參數(shù)是寫入并行度。并行度越高每個(gè)子任務(wù)分到的數(shù)據(jù)越少生成的文件越多但每個(gè)文件越小。我們的經(jīng)驗(yàn)是寫入并行度不要超過Kafka分區(qū)數(shù)否則會(huì)有子任務(wù)空閑。同時(shí)如果發(fā)現(xiàn)生成的文件普遍偏小可以適當(dāng)降低并行度讓每個(gè)子任務(wù)攢更多數(shù)據(jù)再寫。2.4 Schema演進(jìn)策略與兼容性設(shè)計(jì)Schema演進(jìn)是數(shù)據(jù)湖相比傳統(tǒng)數(shù)倉的一大優(yōu)勢但用不好也會(huì)踩坑。Iceberg支持的操作包括新增列、刪除列、重命名列、修改列類型有限制、調(diào)整列順序等。Flink寫入端和下游讀取端對(duì)schema變更的感知是異步的需要設(shè)計(jì)好兼容策略。我們的做法是所有schema變更走統(tǒng)一的元數(shù)據(jù)管理流程。新增字段時(shí)先在Iceberg表上執(zhí)行ALTER TABLE ADD COLUMN然后更新Flink寫入作業(yè)的schema定義最后通知下游讀取方更新。這個(gè)順序不能亂否則寫入端會(huì)報(bào)字段不存在的錯(cuò)誤。對(duì)于列類型修改Iceberg支持int到long、float到double這種安全的類型提升但不支持反向轉(zhuǎn)換。我們規(guī)定只允許類型提升不允許類型收窄。如果業(yè)務(wù)確實(shí)需要收窄類型走新建表加數(shù)據(jù)遷移的流程不在原表上直接改。還有一個(gè)容易忽略的點(diǎn)Flink SQL中定義的schema和Iceberg表的schema要定期做一致性校驗(yàn)。我們遇到過Flink作業(yè)的schema定義和Iceberg表實(shí)際schema不一致的情況原因是有人直接在Iceberg表上加了字段但忘了更新Flink作業(yè)。后來我們加了一個(gè)定時(shí)校驗(yàn)任務(wù)每天對(duì)比Flink作業(yè)的schema和Iceberg表的schema不一致就告警。2.5 小文件治理的自動(dòng)與手動(dòng)策略小文件是流式寫入數(shù)據(jù)湖的頑疾。Iceberg提供了幾種治理手段我們根據(jù)數(shù)據(jù)的重要程度和查詢頻率采用了分層策略。對(duì)于高優(yōu)表核心業(yè)務(wù)表查詢頻繁我們開啟了Iceberg的自動(dòng)合并功能。在Flink寫入端配置write.distribution-mode為hash讓相同分區(qū)鍵的數(shù)據(jù)盡量寫到同一個(gè)文件里。同時(shí)在Iceberg側(cè)配置定期執(zhí)行rewrite_data_files和rewrite_manifests操作把小于目標(biāo)大小的文件合并成大文件清理過期的manifest。對(duì)于普通表日志類、歸檔類我們采用手動(dòng)合并策略每天凌晨低峰期跑一次合并任務(wù)。合并任務(wù)的參數(shù)需要根據(jù)表的數(shù)據(jù)量和查詢模式來調(diào)。比如target-file-size-bytes設(shè)為256MBmin-file-size-bytes設(shè)為64MB小于64MB的文件會(huì)被合并。max-file-size-bytes設(shè)為512MB超過這個(gè)大小的文件不再合并。對(duì)于極冷表很少查詢的歷史歸檔我們不做合并只做定期的snapshot過期清理刪除不再需要的舊snapshot和對(duì)應(yīng)的數(shù)據(jù)文件釋放存儲(chǔ)空間。這里有個(gè)實(shí)操細(xì)節(jié)合并操作本身也會(huì)消耗計(jì)算資源。如果合并任務(wù)和實(shí)時(shí)寫入任務(wù)跑在同一個(gè)集群上可能會(huì)互相影響。我們的做法是把合并任務(wù)調(diào)度到獨(dú)立的批處理集群上通過讀取Iceberg表的snapshot來做合并不影響實(shí)時(shí)寫入鏈路。3. 完整實(shí)操流程與核心環(huán)節(jié)實(shí)現(xiàn)3.1 環(huán)境準(zhǔn)備與依賴版本鎖定在開始集成之前版本兼容性是第一個(gè)要確認(rèn)的事情。Flink和Iceberg的版本對(duì)應(yīng)關(guān)系比較敏感版本不匹配會(huì)導(dǎo)致各種奇怪的報(bào)錯(cuò)。我們當(dāng)時(shí)用的組合是Flink 1.13.2 Iceberg 0.12.0 Hadoop 3.2.1。這個(gè)組合是經(jīng)過社區(qū)驗(yàn)證的穩(wěn)定版本。依賴方面需要在Flink的lib目錄下放入以下jar包# Iceberg核心包 iceberg-flink-runtime-0.12.0.jar # Hive Catalog依賴如果使用HiveCatalog iceberg-hive-runtime-0.12.0.jar # Hadoop依賴如果Flink集群沒有自帶 hadoop-common-3.2.1.jar hadoop-hdfs-client-3.2.1.jar注意iceberg-flink-runtime是一個(gè)shaded包已經(jīng)包含了大部分依賴不要重復(fù)引入Iceberg的其他模塊否則會(huì)出現(xiàn)類沖突。Hive Metastore的配置需要在Flink的sql-client-defaults.yaml或者代碼中指定catalogs: - name: iceberg_catalog type: iceberg catalog-type: hive uri: thrift://your-hive-metastore:9083 warehouse: hdfs://your-namenode:8020/warehouse/iceberg3.2 Flink SQL建表與參數(shù)配置詳解建表是第一步也是參數(shù)配置最集中的地方。下面是一個(gè)典型的upsert模式Iceberg表的建表語句CREATE TABLE iceberg_catalog.db.order_detail ( order_id BIGINT, user_id BIGINT, product_id BIGINT, order_amount DECIMAL(10,2), order_status STRING, create_time TIMESTAMP(3), update_time TIMESTAMP(3), dt STRING, PRIMARY KEY (order_id) NOT ENFORCED ) PARTITIONED BY (dt) WITH ( connector iceberg, catalog-type hive, catalog-database db, catalog-table order_detail, write.distribution-mode hash, write.upsert.enabled true, write.metadata.delete-after-commit.enabled true, write.metadata.previous-versions-max 10, write.format.default parquet, write.parquet.compression-codec zstd, write.target-file-size-bytes 268435456 );幾個(gè)關(guān)鍵參數(shù)的解釋write.distribution-mode設(shè)為hash表示按照分區(qū)鍵做hash分發(fā)相同分區(qū)的數(shù)據(jù)會(huì)寫到同一個(gè)Flink子任務(wù)減少小文件數(shù)量。如果設(shè)為none數(shù)據(jù)隨機(jī)分發(fā)小文件會(huì)更多。write.upsert.enabled設(shè)為true開啟upsert模式。注意這個(gè)參數(shù)在Flink SQL中是通過主鍵定義隱式開啟的如果定義了PRIMARY KEYIceberg會(huì)自動(dòng)使用upsert模式。write.metadata.delete-after-commit.enabled設(shè)為true每次提交后自動(dòng)刪除過期的元數(shù)據(jù)文件防止元數(shù)據(jù)無限膨脹。write.metadata.previous-versions-max設(shè)為10保留最近10個(gè)版本的元數(shù)據(jù)支持時(shí)間旅行查詢最近10個(gè)snapshot。write.target-file-size-bytes設(shè)為256MB這是目標(biāo)文件大小Iceberg在寫入時(shí)會(huì)盡量讓每個(gè)文件接近這個(gè)大小。write.parquet.compression-codec設(shè)為zstd壓縮比和性能比較均衡。如果對(duì)存儲(chǔ)成本敏感可以用zstd如果對(duì)讀取性能要求極高可以用snappy。3.3 實(shí)時(shí)寫入作業(yè)的提交與調(diào)優(yōu)建完表之后提交Flink流作業(yè)寫入數(shù)據(jù)。作業(yè)的并行度設(shè)置需要根據(jù)數(shù)據(jù)量和集群資源來定。我們的經(jīng)驗(yàn)值是每個(gè)并行子任務(wù)的寫入吞吐控制在5MB/s到10MB/s之間。如果總吞吐是100MB/s那么并行度設(shè)為10到20比較合適。Checkpoint的配置在Flink作業(yè)的配置文件中execution.checkpointing.interval: 30s execution.checkpointing.mode: EXACTLY_ONCE execution.checkpointing.timeout: 10min execution.checkpointing.min-pause: 10s execution.checkpointing.max-concurrent-checkpoints: 1 state.backend: rocksdb state.backend.incremental: trueexecution.checkpointing.interval設(shè)為30秒和前面計(jì)算的文件大小目標(biāo)對(duì)應(yīng)。execution.checkpointing.timeout設(shè)為10分鐘給Iceberg提交snapshot留足時(shí)間。如果超時(shí)時(shí)間設(shè)得太短Iceberg提交還沒完成checkpoint就超時(shí)了會(huì)導(dǎo)致作業(yè)頻繁重啟。execution.checkpointing.max-concurrent-checkpoints設(shè)為1禁止并發(fā)checkpoint。Iceberg的snapshot提交需要串行執(zhí)行并發(fā)checkpoint會(huì)導(dǎo)致提交沖突。state.backend用RocksDB開啟增量checkpoint減少狀態(tài)存儲(chǔ)的開銷。3.4 離線讀取與時(shí)間旅行查詢實(shí)操Iceberg表建好、數(shù)據(jù)寫進(jìn)去之后離線讀取就很簡單了。用Flink SQL或者Spark SQL都可以直接查-- Flink SQL查詢最新數(shù)據(jù) SELECT * FROM iceberg_catalog.db.order_detail WHERE dt 2024-01-15; -- Flink SQL時(shí)間旅行查詢查詢某個(gè)snapshot的數(shù)據(jù) SELECT * FROM iceberg_catalog.db.order_detail /* OPTIONS(snapshot-id1234567890) */ WHERE dt 2024-01-15;時(shí)間旅行查詢?cè)谂挪閿?shù)據(jù)問題時(shí)特別有用。比如業(yè)務(wù)方反饋某天的數(shù)據(jù)有問題我們可以通過時(shí)間旅行查詢到那天的snapshot對(duì)比前后幾個(gè)snapshot的數(shù)據(jù)變化快速定位是哪次寫入引入了異常數(shù)據(jù)。還有一個(gè)實(shí)用技巧通過Iceberg的元數(shù)據(jù)表查詢表的文件分布情況。Iceberg提供了snapshots、manifests、files等元數(shù)據(jù)表可以直接用SQL查詢-- 查詢表的所有snapshot SELECT * FROM iceberg_catalog.db.order_detail.snapshots; -- 查詢某個(gè)snapshot的文件列表 SELECT * FROM iceberg_catalog.db.order_detail.files WHERE snapshot_id 1234567890;這些元數(shù)據(jù)表在排查小文件問題、分析數(shù)據(jù)分布時(shí)非常有用。3.5 監(jiān)控指標(biāo)與告警配置生產(chǎn)環(huán)境跑起來之后監(jiān)控是必不可少的。我們主要監(jiān)控以下幾類指標(biāo)Flink作業(yè)層面checkpoint成功率、checkpoint耗時(shí)、背壓情況、吞吐量。這些指標(biāo)通過Flink的Metric系統(tǒng)采集接入Prometheus和Grafana做可視化。Iceberg表層面snapshot數(shù)量、manifest文件數(shù)量、數(shù)據(jù)文件數(shù)量、平均文件大小、元數(shù)據(jù)文件大小。這些指標(biāo)通過定時(shí)查詢Iceberg的元數(shù)據(jù)表來采集。存儲(chǔ)層面HDFS的存儲(chǔ)使用量、文件數(shù)量增長趨勢。如果發(fā)現(xiàn)文件數(shù)量增長過快說明小文件問題嚴(yán)重需要觸發(fā)合并任務(wù)。告警規(guī)則方面我們?cè)O(shè)置了幾個(gè)關(guān)鍵閾值checkpoint連續(xù)失敗3次告警、單次checkpoint耗時(shí)超過5分鐘告警、Iceberg表的小文件數(shù)量小于64MB的文件超過1000個(gè)告警、元數(shù)據(jù)文件大小超過1GB告警。4. 常見問題與排查技巧實(shí)錄4.1 寫入性能突然下降的排查路徑這是我們?cè)谏a(chǎn)環(huán)境遇到最多的問題。表現(xiàn)是Flink作業(yè)的吞吐量突然下降checkpoint耗時(shí)變長背壓升高。排查路徑如下第一步看Flink的背壓情況。通過Flink Web UI或者M(jìn)etric系統(tǒng)查看各個(gè)算子的背壓狀態(tài)。如果背壓出現(xiàn)在Iceberg Sink算子說明寫入Iceberg這一側(cè)有問題。第二步看Iceberg表的文件數(shù)量。如果文件數(shù)量在短時(shí)間內(nèi)快速增長說明小文件問題嚴(yán)重了。大量小文件會(huì)導(dǎo)致Iceberg在提交snapshot時(shí)需要處理大量的manifest文件提交耗時(shí)變長進(jìn)而拖慢整個(gè)checkpoint。第三步看HDFS的寫入延遲。如果HDFS的寫入延遲升高也會(huì)導(dǎo)致Iceberg寫入變慢。這種情況通常是HDFS集群本身的問題比如NameNode負(fù)載過高、DataNode磁盤IO瓶頸等。第四步看Iceberg的提交日志。Iceberg在提交snapshot時(shí)會(huì)打印詳細(xì)的日志包括生成了多少個(gè)數(shù)據(jù)文件、多少個(gè)刪除文件、提交耗時(shí)等。通過日志可以定位到具體的瓶頸。我們遇到過一次典型的性能下降原因是某個(gè)Flink作業(yè)的并行度從10調(diào)到了20導(dǎo)致每個(gè)子任務(wù)寫入的數(shù)據(jù)量減半生成的文件數(shù)量翻倍小文件問題加劇最終拖慢了整個(gè)寫入鏈路。解決辦法是把并行度調(diào)回10同時(shí)開啟自動(dòng)合并。4.2 Checkpoint失敗與數(shù)據(jù)不一致的修復(fù)Checkpoint失敗是流作業(yè)的常見問題但和Iceberg集成后失敗的原因和修復(fù)方式有一些特殊性。場景一Iceberg提交超時(shí)導(dǎo)致checkpoint失敗。表現(xiàn)是checkpoint超時(shí)日志里能看到Iceberg的commit操作還在執(zhí)行。這種情況通常是Iceberg提交的數(shù)據(jù)量太大或者HDFS寫入太慢。解決辦法是縮短checkpoint間隔減少每次提交的數(shù)據(jù)量或者增加checkpoint的超時(shí)時(shí)間。場景二Iceberg提交成功但checkpoint失敗。這種情況比較棘手因?yàn)镮ceberg已經(jīng)提交了snapshot但Flink認(rèn)為checkpoint失敗了會(huì)回滾到上一個(gè)checkpoint。這會(huì)導(dǎo)致Iceberg表里多了一批“孤兒數(shù)據(jù)”雖然不影響讀取因?yàn)樽x取是基于snapshot的但會(huì)浪費(fèi)存儲(chǔ)空間。解決辦法是配置Iceberg的write.metadata.delete-after-commit.enabled為true定期清理孤兒snapshot。場景三Checkpoint成功但I(xiàn)ceberg提交失敗。這種情況會(huì)導(dǎo)致數(shù)據(jù)丟失因?yàn)镕link認(rèn)為checkpoint成功了不會(huì)重放這部分?jǐn)?shù)據(jù)但I(xiàn)ceberg里沒有寫入。這種情況通常是由于Iceberg的Catalog不可用或者權(quán)限問題導(dǎo)致的。解決辦法是配置重試機(jī)制Iceberg的commit操作支持重試可以在Flink的Sink配置中設(shè)置重試次數(shù)和重試間隔。實(shí)操心得我們后來在Flink作業(yè)里加了一個(gè)“提交確認(rèn)”機(jī)制每次checkpoint完成后額外做一次Iceberg的snapshot校驗(yàn)確認(rèn)最新的snapshot ID和checkpoint的ID對(duì)應(yīng)。如果不一致觸發(fā)告警并暫停作業(yè)人工介入排查。4.3 小文件過多的緊急處理方案小文件問題如果已經(jīng)積累到影響查詢性能的程度需要緊急處理。我們的處理流程是第一步評(píng)估影響范圍。查詢Iceberg的files元數(shù)據(jù)表統(tǒng)計(jì)小于64MB的文件數(shù)量和總大小評(píng)估合并需要的時(shí)間和資源。第二步暫停實(shí)時(shí)寫入。如果小文件問題非常嚴(yán)重合并任務(wù)和實(shí)時(shí)寫入同時(shí)跑會(huì)互相影響建議先暫停實(shí)時(shí)寫入作業(yè)等合并完成后再恢復(fù)。暫停時(shí)間根據(jù)合并的數(shù)據(jù)量來定一般幾十分鐘到幾個(gè)小時(shí)。第三步執(zhí)行合并任務(wù)。用Spark或者Flink批作業(yè)執(zhí)行Iceberg的rewrite_data_files操作-- Spark SQL執(zhí)行合并 CALL iceberg_catalog.system.rewrite_data_files( table db.order_detail, options map( target-file-size-bytes, 268435456, min-file-size-bytes, 67108864, max-file-size-bytes, 536870912 ) );第四步清理過期元數(shù)據(jù)。合并完成后執(zhí)行rewrite_manifests和expire_snapshots清理不再需要的manifest文件和snapshot。第五步恢復(fù)實(shí)時(shí)寫入。確認(rèn)合并完成后恢復(fù)Flink作業(yè)并調(diào)整參數(shù)防止小文件問題再次出現(xiàn)。4.4 Schema變更導(dǎo)致作業(yè)失敗的應(yīng)急處理Schema變更導(dǎo)致的作業(yè)失敗通常發(fā)生在寫入端和讀取端的schema不一致時(shí)。應(yīng)急處理流程情況一寫入端報(bào)字段不存在。原因是Iceberg表加了字段但Flink作業(yè)的schema沒更新。解決辦法是更新Flink作業(yè)的schema定義并重啟作業(yè)。如果作業(yè)不能立即重啟可以先用ALTER TABLE DROP COLUMN把新加的字段刪掉等作業(yè)更新后再加回來。情況二讀取端報(bào)字段類型不匹配。原因是Iceberg表的字段類型改了但讀取端的schema沒更新。解決辦法是更新讀取端的schema定義。如果讀取端是BI工具或者即席查詢需要通知相關(guān)方更新。情況三寫入端報(bào)主鍵沖突。原因是upsert模式的主鍵定義和Iceberg表的主鍵定義不一致。解決辦法是檢查兩邊的定義確保完全一致。避坑技巧我們后來規(guī)定所有schema變更必須走工單流程變更前先在測試環(huán)境驗(yàn)證確認(rèn)寫入端和讀取端都能正常工作后再上生產(chǎn)。同時(shí)在Flink作業(yè)里加了schema兼容性檢查作業(yè)啟動(dòng)時(shí)對(duì)比Flink schema和Iceberg schema不一致直接拒絕啟動(dòng)。4.5 常見問題速查表問題現(xiàn)象可能原因排查方法解決方案寫入吞吐突然下降小文件過多、HDFS延遲高查看Iceberg文件數(shù)量、HDFS監(jiān)控執(zhí)行合并、調(diào)整并行度Checkpoint頻繁超時(shí)單次提交數(shù)據(jù)量太大查看checkpoint耗時(shí)、Iceberg提交日志縮短checkpoint間隔、增加超時(shí)時(shí)間數(shù)據(jù)重復(fù)upsert模式未生效檢查主鍵定義、查詢Iceberg表數(shù)據(jù)確認(rèn)主鍵定義、開啟upsert查詢性能差分區(qū)裁剪失效、文件過大查看查詢計(jì)劃、文件大小分布調(diào)整分區(qū)策略、合并大文件元數(shù)據(jù)膨脹snapshot未清理查看snapshots元數(shù)據(jù)表配置自動(dòng)清理、執(zhí)行expire_snapshots作業(yè)啟動(dòng)失敗schema不一致、依賴沖突查看啟動(dòng)日志、對(duì)比schema更新schema、檢查jar包版本時(shí)間旅行查詢報(bào)錯(cuò)snapshot已過期查看snapshots元數(shù)據(jù)表調(diào)整snapshot保留策略5. 生產(chǎn)環(huán)境運(yùn)維經(jīng)驗(yàn)與性能優(yōu)化心得5.1 資源規(guī)劃與容量評(píng)估方法Iceberg表的存儲(chǔ)容量增長是可以預(yù)估的。核心公式是日增數(shù)據(jù)量 × 副本數(shù) × 壓縮比 × 保留天數(shù)。比如日增原始數(shù)據(jù)100GBHDFS副本數(shù)3Parquet壓縮比0.3保留365天那么總存儲(chǔ)需求大約是100GB × 3 × 0.3 × 365 ≈ 32TB。這還不包括合并操作產(chǎn)生的臨時(shí)文件和舊snapshot占用的空間實(shí)際規(guī)劃時(shí)要留出30%到50%的余量。計(jì)算資源方面Flink寫入作業(yè)的資源需求主要取決于寫入吞吐和checkpoint的數(shù)據(jù)量。我們的經(jīng)驗(yàn)值是每10MB/s的寫入吞吐需要大約4個(gè)CPU核心和8GB內(nèi)存。這個(gè)值會(huì)根據(jù)數(shù)據(jù)復(fù)雜度、序列化開銷、Iceberg提交開銷等因素浮動(dòng)建議在實(shí)際壓測后確定。合并任務(wù)的資源需求取決于表的數(shù)據(jù)量和合并頻率。每天合并一次的話合并任務(wù)的處理能力建議是日增數(shù)據(jù)量的2到3倍確保能在低峰期內(nèi)完成合并。5.2 多表共用一個(gè)Flink作業(yè)的利弊分析為了節(jié)省資源我們一開始嘗試把多個(gè)Iceberg表的寫入放在同一個(gè)Flink作業(yè)里。好處是節(jié)省了TaskManager的資源減少了作業(yè)數(shù)量運(yùn)維成本低。但壞處也很明顯故障隔離性差。一張表的寫入出現(xiàn)問題整個(gè)作業(yè)都會(huì)受影響所有表的數(shù)據(jù)寫入都會(huì)延遲。資源競爭。不同表的寫入吞吐不同有的表數(shù)據(jù)量大、有的表數(shù)據(jù)量小共用一個(gè)作業(yè)會(huì)導(dǎo)致資源分配不均數(shù)據(jù)量大的表可能搶不到足夠的資源。Checkpoint相互影響。一個(gè)作業(yè)的checkpoint是全局的如果某張表的Iceberg提交特別慢會(huì)拖慢整個(gè)checkpoint影響所有表。后來我們改成了按業(yè)務(wù)域拆分作業(yè)每個(gè)業(yè)務(wù)域的核心表放在一個(gè)作業(yè)里非核心表單獨(dú)跑。這樣既保證了核心表的穩(wěn)定性又控制了作業(yè)數(shù)量。5.3 數(shù)據(jù)回溯與重跑的正確姿勢數(shù)據(jù)回溯是數(shù)據(jù)平臺(tái)的剛需Iceberg在這方面提供了很好的支持。我們的回溯流程是第一步確定回溯范圍。通過時(shí)間旅行查詢確定要回溯到哪個(gè)snapshot以及需要重跑的時(shí)間段。第二步暫停下游消費(fèi)。通知下游停止消費(fèi)Iceberg表的數(shù)據(jù)避免讀到中間狀態(tài)的數(shù)據(jù)。第三步執(zhí)行回溯寫入。用Flink批作業(yè)或者Spark作業(yè)重新計(jì)算指定時(shí)間段的數(shù)據(jù)寫入Iceberg表。注意寫入時(shí)要指定新的分區(qū)或者使用INSERT OVERWRITE語義避免和現(xiàn)有數(shù)據(jù)沖突。第四步校驗(yàn)數(shù)據(jù)?；厮萃瓿珊髮?duì)比回溯前后的數(shù)據(jù)量和關(guān)鍵指標(biāo)確認(rèn)數(shù)據(jù)正確。第五步恢復(fù)下游消費(fèi)。通知下游恢復(fù)消費(fèi)并告知回溯完成的時(shí)間點(diǎn)。實(shí)操心得回溯操作最好在低峰期執(zhí)行并且提前通知相關(guān)方。我們有一次在業(yè)務(wù)高峰期做回溯導(dǎo)致Iceberg表的寫入壓力驟增影響了實(shí)時(shí)鏈路的性能。后來規(guī)定所有回溯操作必須提前一天報(bào)備安排在凌晨執(zhí)行。5.4 從Iceberg 0.12到0.14的升級(jí)踩坑記錄Iceberg社區(qū)迭代很快我們后來從0.12升級(jí)到了0.14中間踩了一些坑記錄一下坑一Flink版本兼容性。Iceberg 0.14要求Flink 1.14及以上版本我們當(dāng)時(shí)用的Flink 1.13.2不兼容需要先升級(jí)Flink。升級(jí)Flink又涉及到狀態(tài)兼容性問題需要做狀態(tài)遷移?？佣嗀PI變更。Iceberg 0.14對(duì)Flink SQL的Connector參數(shù)做了一些調(diào)整比如write.upsert.enabled參數(shù)被廢棄改為通過主鍵定義自動(dòng)識(shí)別。升級(jí)后需要修改建表語句?？尤獢?shù)據(jù)格式變更。Iceberg 0.14對(duì)元數(shù)據(jù)格式做了一些優(yōu)化升級(jí)后需要執(zhí)行一次元數(shù)據(jù)遷移操作。這個(gè)操作是不可逆的升級(jí)前一定要備份元數(shù)據(jù)。坑四性能表現(xiàn)差異。升級(jí)后發(fā)現(xiàn)寫入性能比0.12略有下降排查后發(fā)現(xiàn)是0.14默認(rèn)開啟了更嚴(yán)格的元數(shù)據(jù)校驗(yàn)。通過調(diào)整write.metadata.compression-codec參數(shù)把元數(shù)據(jù)壓縮算法從gzip改為zstd性能恢復(fù)到了升級(jí)前的水平。5.5 未來演進(jìn)方向與擴(kuò)展思路Iceberg和Flink的集成還在快速演進(jìn)中我們也在持續(xù)關(guān)注幾個(gè)方向一是Flink CDC與Iceberg的深度集成。目前我們是從Kafka接binlog數(shù)據(jù)寫入Iceberg鏈路比較長。Flink CDC可以直接從數(shù)據(jù)庫抓取變更數(shù)據(jù)寫入Iceberg鏈路更短、延遲更低。我們正在測試這個(gè)方案初步效果不錯(cuò)。二是Iceberg的Row-level Delete優(yōu)化。Iceberg 0.14引入了更高效的刪除文件格式upsert模式的寫入性能有明顯提升。我們計(jì)劃在下一個(gè)版本升級(jí)中驗(yàn)證這個(gè)特性。三是多引擎共享Iceberg表。目前我們主要是Flink寫入、Flink和Spark讀取。未來計(jì)劃讓Trino、Presto等引擎也能直接查詢Iceberg表實(shí)現(xiàn)真正的多引擎共享。四是冷熱數(shù)據(jù)分層存儲(chǔ)。Iceberg支持自定義文件存儲(chǔ)路徑可以把冷數(shù)據(jù)放到更便宜的存儲(chǔ)介質(zhì)上熱數(shù)據(jù)保留在高性能存儲(chǔ)上。這個(gè)特性對(duì)降低存儲(chǔ)成本很有幫助我們正在做方案設(shè)計(jì)。整體來看Flink加Iceberg的組合在同程藝龍的實(shí)踐是成功的。流批一體不是一句口號(hào)而是實(shí)實(shí)在在解決了我們數(shù)據(jù)口徑不一致、鏈路重復(fù)建設(shè)的問題。當(dāng)然落地過程中遇到的坑也不少上面這些經(jīng)驗(yàn)都是一個(gè)個(gè)問題踩出來的。如果你也在做類似的實(shí)踐希望這些內(nèi)容能幫你少走一些彎路。