Apache Iceberg Spark Streaming 构建 Lakehouse 实时数仓CDC 增量入湖与查询加速实践摘要传统 Lambda 架构在实时数仓建设中面临数据一致性与维护成本的双重挑战。本文以 Apache Iceberg 表格式为核心结合 Spark Structured Streaming 与 CDC变更数据捕获技术详细阐述如何构建一套支撑分钟级延迟的 Lakehouse 实时数仓。内容涵盖 Iceberg 元数据管理机制、Spark-Iceberg 增量读写原理、Flink CDC 整库同步方案以及查询层的分区裁剪与文件编排优化提供可直接落地的代码与配置。一、Lambda 架构的困境与 Lakehouse 的破局在过去五年的数据平台建设中我们团队一直沿用经典的 Lambda 架构批处理层Spark 每日 T1 处理 Hive 数据生成历史全量报表速度层Flink 实时消费 Kafka写入 HBase / ClickHouse 供实时查询服务层合并批与流的结果暴露给 BI 工具。这套架构的问题在数据规模突破 PB 级后集中爆发同一业务逻辑需要维护批流两套代码数据口径不一致引发的数字对不上成为分析师的日常噩梦。更严重的是HBase 的 Schema 变更几乎等同于停服重建无法满足业务快速迭代的需要。Lakehouse 架构的提出核心在于用开放的表格式Apache Iceberg / Hudi / Delta Lake在对象存储之上实现数仓的 ACID 语义。其中 Apache Iceberg 凭借其纯开源、无 vendor lock-in不绑定特定计算引擎优秀的生态系统Spark、Flink、Trino、StarRocks 均有成熟连接器先进的元数据设计隐式分区、Time-Travel、Partition Evolution成为我们在 2024-2025 年技术升级的首选。二、Iceberg 元数据架构理解隐式分区的钥匙很多工程师初次接触 Iceberg 时会困惑于为什么查询时不需要指定分区字段。要回答这个问题必须深入理解 Iceberg 的三层元数据模型。2.1 Catalog → Table → Snapshot → Manifest → DataFileIceberg 的元数据分为以下层级CatalogHive / Hadoop / JDBC / REST └── Table Metadata JSON └── Snapshot List多版本快照 └── Manifest List └── Manifest File分区统计信息 DataFile 列表 └── DataFileParquet / ORC / Avro关键设计每个 Snapshot 是一个不可变的表状态。当执行INSERT、UPDATE或DELETE时Iceberg 不会修改任何已有 DataFile而是写入新的 DataFile并在 Manifest 中记录文件的 min/max 统计信息最后生成一个新的 Snapshot 并切换current-snapshot-id。这意味着Time-Travel天然支持SELECT * FROM table TIMESTAMP AS OF 2025-06-01 10:00:00只需回溯到对应 Snapshot并发写入安全乐观锁机制下两个 Spark Job 同时提交时后提交的 Job 会检测到元数据版本变化并自动重试分区演进无痛修改分区策略不会影响历史数据新数据按新分区写入查询时引擎自动选择最优裁剪策略。2.2 隐式分区与查询裁剪传统 Hive 表需要用户显式匹配分区字段如WHERE dt2025-08-11而 Iceberg 在 Manifest 文件中记录了每个 DataFile 的列级统计信息min/max、null count、distinct values。当查询带有过滤条件时Iceberg 通过底层 API 的planFiles()方法在读取任何 Parquet 文件之前先根据统计信息过滤掉不相关的 DataFile。在我们的生产环境中一张 500 亿行的用户行为表通过 Iceberg 的隐式分区 Z-Order 排序将全表扫描查询的 IO 量降低了 97%。三、CDC 增量入湖Flink CDC Iceberg 整库同步实时数仓的核心是数据新鲜度。我们需要将 MySQL、Oracle 等业务库的变更实时捕获并写入 Iceberg。3.1 技术选型Flink CDC 3.0Flink CDC 3.0 引入了整库同步能力支持 Schema Evolution加列、改类型自动同步到下游 Iceberg 表。相比早期版本需要为每张表单独写 Flink Job3.0 版本只需一个 YAML 配置文件即可同步整库。3.2 YAML 配置实战以下是我们同步核心订单库的配置source:type:mysqlhostname:mysql-primary.db.svcport:3306username:cdc_userpassword:${CDC_PASSWORD}tables:order_db.order_info,order_db.order_detailserver-id:5400-5404sink:type:icebergcatalog:type:hadoopwarehouse:hdfs://namenode:8020/warehouse/icebergtable-prefix:cdc_table-defaults:format-version:2write.metadata.metrics.default:counts# 默认收集列统计write.distribution-mode:hash# 按主键哈希分布避免小文件pipeline:parallelism:8schema-change-mode:evolve# 自动同步 Schema 变更关键参数解析format-version: 2Iceberg V2 表格式支持行级 UPDATE/DELETE基于 position/equality delete files是 CDC 场景的必要条件write.distribution-mode: hash确保同一主键的数据落在同一文件减少后续 Merge-on-Read 时的文件扫描范围schema-change-mode: evolve当上游 MySQL 执行ALTER TABLE ADD COLUMN时Flink CDC 自动在 Iceberg 表上执行对应变更无需人工介入。3.3 写入优化WAL 与 Checkpoint 调优Flink CDC 写入 Iceberg 时每个 Checkpoint 会触发一次 Iceberg Commit。如果 Checkpoint 间隔过短如 1 秒会产生大量小文件和元数据膨胀如果过长如 10 分钟数据延迟又会超标。我们的调优策略是分层 CheckpointStreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(30000);// Checkpoint 30 秒平衡延迟与小文件env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);env.getCheckpointConfig().setMinPauseBetweenCheckpoints(20000);配合 Iceberg 的write.target-file-size-bytes134217728128MB使每个 DataFile 大小稳定在 100-150MB 之间兼顾查询效率与写入吞吐。3.4 Merge-on-Read vs Copy-on-WriteIceberg V2 提供两种更新模式模式写入路径读取路径适用场景Copy-on-WriteCOW重写整个 DataFile直接读 DataFile写少读多Merge-on-ReadMOR写 Delete File 新 DataFile合并读取写多读少CDC 场景下由于变更频率高我们选择MOR 模式。读取时通过read.delete.modemerge-on-read配置Iceberg 会自动将 Delete File 中的记录排除。对于准实时报表我们额外配置了 Spark 定时 Compaction Job每小时一次将 Delete File 合并到 DataFile 中避免读放大。四、Spark Structured Streaming流式增量计算数据入湖后需要在 Iceberg 之上进行增量 ETL生成 DWD明细层和 DWS汇总层。Spark Structured Streaming 与 Iceberg 的集成提供了微批Micro-batch和连续处理Continuous Processing两种模式。4.1 增量读取原理Spark 通过 Iceberg 的IncrementalChangelogScanAPI 实现增量读取。每次微批启动时Spark 查询自上次 Checkpoint 以来新增的所有 Snapshot只读取新增的 DataFile。valdfspark.readStream.format(iceberg).option(stream-from-timestamp,startTimestamp).load(warehouse.iceberg.cdc_order_info)valquerydf.writeStream.format(iceberg).outputMode(append).option(checkpointLocation,/checkpoints/order_dwd).toTable(warehouse.iceberg.dwd_order_event)4.2 DWD 层构建事件清洗与打宽在 DWD 层我们需要将订单主表与详情表关联并补充用户维度信息。由于 Iceberg 支持 ACID我们可以使用 MERGE INTO 实现幂等的 UpsertMERGEINTOwarehouse.iceberg.dwd_order_event tUSING(SELECT*FROMstreaming_batch)sONt.order_ids.order_idWHENMATCHEDTHENUPDATESET*WHENNOTMATCHEDTHENINSERT*关键优化点Broadcast Hint维度表如用户表仅 200MB通过/* BROADCAST(dim_user) */强制广播避免 ShuffleZ-Order 排序对 DWD 表执行OPTIMIZE table ZORDER BY (user_id, event_time)将同一用户的数据聚类到相邻文件极大加速后续用户级聚合查询。4.3 窗口聚合与 WatermarkDWS 层需要按 5 分钟滚动窗口统计订单金额。Structured Streaming 的 Watermark 机制用于处理乱序数据valwindowedCountsdf.withWatermark(event_time,10 minutes).groupBy(window($event_time,5 minutes),$region).agg(sum($amount).as(total_amount))WaterMark 延迟设为 10 分钟意味着 10 分钟前的窗口会被触发并写入 Iceberg。由于 Iceberg 的 Snapshot 隔离性下游查询不会读到未闭合的窗口数据保证了读到即完整的语义。五、查询加速分区演进、隐藏分区与文件编排5.1 分区演进Partition Evolution业务初期订单表按days(order_time)分区即可满足需求。随着数据量增长我们发现同一分区内的文件过多每日 10 万 文件查询启动时的文件列表耗时成为瓶颈。Iceberg 支持分区演进在不重建表的情况下修改分区策略使新数据按更细的粒度分区。-- 原始分区策略按天ALTERTABLEorder_infoADDPARTITIONFIELD hours(order_time);执行后历史数据仍按天组织新写入数据按小时组织。查询引擎根据时间范围自动选择最优的分区粒度进行裁剪。5.2 隐藏分区Hidden Partitioning传统 Hive 表中分区字段必须是表中的显式列如dt STRING导致业务 SQL 中充斥WHERE dt2025-08-11这类与业务无关的过滤条件。Iceberg 的隐藏分区允许从现有列派生分区而无需添加冗余列CREATETABLEwarehouse.iceberg.order_info(order_idBIGINT,user_idBIGINT,order_timeTIMESTAMP,amountDECIMAL(16,2))USINGiceberg PARTITIONEDBY(days(order_time),bucket(16,user_id));这里days(order_time)是隐藏分区业务查询只需写WHERE order_time 2025-08-01Iceberg 自动将条件转换为分区过滤。5.3 文件编排OPTIMIZE 与 REWRITE DATACDC 持续写入会产生大量小文件严重影响查询性能。我们通过 Spark 定时作业进行文件编排-- 合并小文件目标 128MBOPTIMIZEwarehouse.iceberg.cdc_order_info;-- Z-Order 重排加速多维过滤REWRITEDATATABLEwarehouse.iceberg.dwd_order_eventUSINGZORDER(user_id,product_id);-- 清理过期 Snapshot释放存储VACUUM warehouse.iceberg.dwd_order_event;生产环境中我们将上述 SQL 封装为 Airflow DAG每日凌晨 2:00 执行将前一天的小文件合并后查询 P95 耗时从 45 秒降至 3 秒。六、查询层集成Trino / StarRocks 统一查询入口湖仓的价值最终体现在查询层。我们在 Iceberg 之上搭建了统一的查询网关Ad-hoc 查询Trino 连接 Iceberg Catalog分析师通过 SQL 直接探查原始数据高并发报表StarRocks 3.x 支持 Iceberg 外表查询通过 Data Cache 将热数据缓存到本地 SSDQPS 可达 5000湖仓一体加速对于查询频率极高的 DWS 汇总表通过 StarRocks 的CREATE MATERIALIZED VIEW将 Iceberg 数据异步导入内表实现亚秒级响应。StarRocks 查询 Iceberg 的关键配置CREATEEXTERNAL RESOURCE iceberg_resource PROPERTIES(typeiceberg,iceberg.catalog.typeHIVE,hive.metastore.uristhrift://hive-metastore:9083);CREATEEXTERNALTABLEext_order_info(order_idBIGINT,amountDECIMAL(16,2))ENGINEICEBERG PROPERTIES(resourceiceberg_resource,databasewarehouse,tabledwd_order_event);StarRocks 的 CBOCost-Based Optimizer会自动将过滤条件下推到 Iceberg利用 Manifest 层的统计信息跳过不满足条件的文件实现与原生数仓表接近的查询性能。七、总结本文围绕 Apache Iceberg Spark Structured Streaming 的技术组合系统阐述了 Lakehouse 实时数仓的构建路径数据入湖利用 Flink CDC 3.0 实现 MySQL 整库分钟级同步借助 Iceberg V2 的 MOR 模式支撑高频更新分层计算Spark Structured Streaming 读取 Iceberg 增量 Snapshot通过 MERGE INTO 构建 DWD/DWSWatermark 机制保证窗口完整性查询加速分区演进、隐藏分区、Z-Order 排序与定时文件编排层层削减查询 IO统一查询Trino 负责灵活探查StarRocks 负责高并发加速实现一份数据、多种负载。相比传统 Lambda 架构该方案将数据链路维护成本降低了约 60%数据一致性达到 Snapshot 隔离级别。随着 Iceberg Spec V4 的推进列式元数据、更快 Commit以及 Paimon 在实时更新场景的持续演进Lakehouse 正在成为实时数仓的事实标准。参考资料Apache Iceberg 官方文档Table Spec PartitioningFlink CDC 3.0 官方文档Pipeline YAML 配置StarRocks 官方文档Iceberg 外表查询Dremio Blog: Looking back the last year in Lakehouse OSS (2025)