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