# Iceberg 数据湖实战

Iceberg 数据湖实战下一代数据湖存储架构系列新技术实战系列难度⭐⭐⭐⭐⭐适合人群5 年 大数据工程师、数据平台架构师前置知识Hadoop 生态、数据仓库概念、Spark/Flink一、为什么需要 Iceberg1.1 传统数据湖的痛点痛点一数据可靠性差场景ETL 作业失败数据只写了一半 传统 Hive 表 - 部分写入的数据可见 - 下游作业读到脏数据 - 需要手动清理和修复 Iceberg: - 原子性提交要么全有要么全无 - 失败自动回滚 - 下游永远读到一致数据痛点二无法更新删除场景GDPR 要求删除用户数据 传统 Hive 表 - 不支持 UPDATE/DELETE - 只能重写整个分区 - 成本高、效率低 Iceberg: - 支持行级 UPDATE/DELETE - Merge-on-Read 高效合并 - 轻松满足合规要求痛点三时间旅行缺失场景数据出错需要回滚到昨天 传统 Hive 表 - 无版本概念 - 需要手动备份 - 回滚复杂 Iceberg: - 内置快照机制 - 支持时间旅行查询 - 一键回滚到任意版本痛点四小文件问题场景Flink 实时写入产生大量小文件 传统 Hive 表 - 小文件拖慢查询 - 需要定期 Compaction - 手动维护 Iceberg: - 自动小文件合并 - 后台优化 - 对查询透明1.2 Iceberg 的核心优势Iceberg Hive 表的升级版 核心优势: 1. ACID 事务 - 原子性、一致性、隔离性、持久性 2. 模式演进 - 添加/删除/重命名列无需重写数据 3. 隐藏分区 - 分区对查询透明自动优化 4. 时间旅行 - 查询历史版本数据 5. 行级操作 - 支持 UPDATE/DELETE/MERGE 6. 多引擎支持 - Spark、Flink、Trino、Hive量化对比特性Hive 表Iceberg 表ACID 事务❌✅行级更新❌✅模式演进困难简单时间旅行❌✅小文件优化手动自动隐藏分区❌✅多版本并发❌✅二、Iceberg 架构原理2.1 三层元数据架构┌─────────────────────────────────────────────────────────────────┐ │ Manifest List │ │ (快照级别的元数据记录本次提交的所有 Manifest File) │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Manifest File (Avro) │ │ (分区级别的元数据记录该分区下所有 Data File 的统计信息) │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Data File (Parquet/ORC) │ │ (实际数据文件包含列统计信息用于谓词下推) │ └─────────────────────────────────────────────────────────────────┘元数据层级说明1. Catalog (目录层) - 记录表的位置和元数据 - 支持 Hive Catalog、Hadoop Catalog、REST Catalog 2. Metadata File (元数据文件) - 表的 Schema、分区、快照历史 - 每次 Schema 变更或快照提交都会生成新文件 3. Manifest List (清单列表) - 每个快照对应一个 Manifest List - 记录该快照包含的所有 Manifest File 4. Manifest File (清单文件) - 记录 Data File 的路径、统计信息 - 用于查询优化谓词下推 5. Data File (数据文件) - Parquet/ORC 格式 - 包含列的最小值、最大值、空值统计2.2 快照模型快照 (Snapshot) 表在某个时间点的状态 快照组成: - Snapshot ID (唯一标识) - Timestamp (时间戳) - Manifest List (数据清单) - Parent Snapshot ID (父快照形成快照链) - Operation (操作类型APPEND/REPLACE/DELETE) 快照链: Snapshot 1 → Snapshot 2 → Snapshot 3 → ... → Snapshot N ↓ ↓ ↓ ↓ 10:00 11:00 12:00 15:00时间旅行查询-- 查询历史版本SELECT*FROMiceberg_tableTIMESTAMPASOF2026-04-08 10:00:00;-- 或指定快照 IDSELECT*FROMiceberg_table VERSIONASOF1234567890;2.3 写入模型Copy-on-Write (CoW)适用场景读多写少 写入流程: 1. 读取受影响的 Data File 2. 在内存中合并新旧数据 3. 写入新的 Data File 4. 更新 Manifest 优点 - 查询性能好无需合并 - 读取路径简单 缺点 - 写入放大需要重写整个文件 - 适合小批量更新Merge-on-Read (MoR)适用场景写多读少实时场景 写入流程: 1. 将变更写入 Delta File追加写 2. 异步 Compaction 合并 3. 查询时合并读取 优点 - 写入快追加写 - 适合流式写入 缺点 - 查询需要合并 - 读取路径复杂三、Iceberg 实战操作3.1 Spark 集成环境配置# Spark 3.3 内置 Iceberg 支持# 或使用独立包spark-submit\--packagesorg.apache.iceberg:iceberg-spark-runtime-3.4_2.12:1.4.0\--confspark.sql.extensionsorg.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions\--confspark.sql.catalog.prodorg.apache.iceberg.spark.SparkCatalog\--confspark.sql.catalog.prod.typehive\--confspark.sql.catalog.prod.urithrift://hive-metastore:9083\my_job.py创建表frompyspark.sqlimportSparkSession sparkSparkSession.builder \.appName(Iceberg Demo)\.config(spark.sql.extensions,org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions)\.config(spark.sql.catalog.prod,org.apache.iceberg.spark.SparkCatalog)\.config(spark.sql.catalog.prod.type,hive)\.getOrCreate()# 创建 Iceberg 表spark.sql( CREATE TABLE prod.iceberg_db.order_detail ( order_id BIGINT COMMENT 订单 ID, user_id BIGINT COMMENT 用户 ID, amount DECIMAL(18,2) COMMENT 订单金额, status INT COMMENT 订单状态, create_time TIMESTAMP COMMENT 创建时间, update_time TIMESTAMP COMMENT 更新时间 ) COMMENT 订单明细表 PARTITIONED BY (days(create_time)) LOCATION s3://data-lake/iceberg/prod/iceberg_db/order_detail TBLPROPERTIES ( write.delete.mode merge-on-read, write.update.mode merge-on-read, write.merge.mode merge-on-read, format-version 2 ) )数据写入# 插入数据spark.sql( INSERT INTO prod.iceberg_db.order_detail SELECT order_id, user_id, amount, status, create_time, update_time FROM ods.order_info WHERE dt 2026-04-08 )# 更新数据spark.sql( UPDATE prod.iceberg_db.order_detail SET status 5, update_time current_timestamp() WHERE order_id 123456789 )# 删除数据spark.sql( DELETE FROM prod.iceberg_db.order_detail WHERE user_id 987654321 )# 合并数据UPSERTspark.sql( MERGE INTO prod.iceberg_db.order_detail AS target USING ( SELECT order_id, user_id, amount, status, create_time, current_timestamp() AS update_time FROM ods.order_incremental ) AS source ON target.order_id source.order_id WHEN MATCHED THEN UPDATE SET target.user_id source.user_id, target.amount source.amount, target.status source.status, target.update_time source.update_time WHEN NOT MATCHED THEN INSERT * )时间旅行查询# 查询历史版本spark.sql( SELECT * FROM prod.iceberg_db.order_detail TIMESTAMP AS OF 2026-04-08 10:00:00 LIMIT 100 )# 查询指定快照spark.sql( SELECT * FROM prod.iceberg_db.order_detail VERSION AS OF 1234567890 LIMIT 100 )# 查看快照历史spark.sql(SELECT * FROM prod.iceberg_db.order_detail.history).show()# 查看文件列表spark.sql(SELECT * FROM prod.iceberg_db.order_detail.files).show()模式演进# 添加新列无需重写数据spark.sql( ALTER TABLE prod.iceberg_db.order_detail ADD COLUMN channel STRING COMMENT 渠道来源 )# 删除列spark.sql( ALTER TABLE prod.iceberg_db.order_detail DROP COLUMN channel )# 重命名列spark.sql( ALTER TABLE prod.iceberg_db.order_detail RENAME COLUMN user_id TO customer_id )# 修改列类型兼容类型spark.sql( ALTER TABLE prod.iceberg_db.order_detail ALTER COLUMN amount TYPE DECIMAL(20,2) )3.2 Flink 集成环境配置# Flink 1.14 支持 Iceberg# 下载 Iceberg Flink 运行时wgethttps://repo1.maven.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.17/1.4.0/iceberg-flink-runtime-1.17-1.4.0.jar# 放入 Flink lib 目录# cp iceberg-flink-runtime-1.17-1.4.0.jar ${FLINK_HOME}/lib/Flink SQL 写入-- Flink SQL Client-- 创建 CatalogCREATECATALOG prodWITH(typeiceberg,catalog-typehive,urithrift://hive-metastore:9083,warehouses3://data-lake/iceberg);-- 创建表CREATETABLEprod.iceberg_db.order_detail(order_idBIGINT,user_idBIGINT,amountDECIMAL(18,2),statusINT,create_timeTIMESTAMP(3),update_timeTIMESTAMP(3),WATERMARKFORcreate_timeAScreate_time-INTERVAL5SECOND)PARTITIONEDBY(days(create_time))WITH(write.delete.modemerge-on-read,write.update.modemerge-on-read,write.merge.modemerge-on-read);-- 实时写入INSERTINTOprod.iceberg_db.order_detailSELECTorder_id,user_id,amount,status,create_time,update_timeFROMkafka_order_source;Flink DataStream API// IcebergDataStreamWriter.javaimportorg.apache.iceberg.flink.sink.FlinkSink;publicclassIcebergDataStreamWriter{publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();// 读取 Kafka 数据DataStreamOrderorderStreamenv.addSource(newFlinkKafkaConsumer(orders,newOrderSchema(),props)).name(kafka-source);// 写入 IcebergTableLoadertableLoaderTableLoader.fromCatalog(CatalogLoader.hive(prod,hiveConf),TableIdentifier.of(prod,iceberg_db,order_detail));FlinkSink.forRowData(orderStream).tableLoader(tableLoader).tableSchema(schema).writeParallelism(4).append();env.execute(Iceberg Realtime Writer);}}3.3 Trino 查询Trino 配置# etc/catalog/iceberg.properties connector.nameiceberg iceberg.catalog.typehive_metastore iceberg.hive.metastore.urithrift://hive-metastore:9083 iceberg.hive.metastore-cache-ttl0s iceberg.hive.metastore-refresh-interval1mTrino 查询-- 查询 Iceberg 表SELECT*FROMiceberg.prod.iceberg_db.order_detailWHEREcreate_timeDATE2026-04-01LIMIT100;-- 时间旅行SELECT*FROMiceberg.prod.iceberg_db.order_detailFORTIMESTAMPASOFTIMESTAMP2026-04-08 10:00:00;-- 查看快照SELECT*FROMiceberg.prod.iceberg_db.order_detail$snapshots;-- 查看历史SELECT*FROMiceberg.prod.iceberg_db.order_detail$history;四、生产环境最佳实践4.1 表设计分区策略-- 推荐按时间分区天/小时PARTITIONEDBY(days(create_time))PARTITIONEDBY(hours(create_time))-- 避免过度分区产生太多小文件PARTITIONEDBY(create_time)-- 错误按时间戳分区-- 推荐组合分区时间 业务PARTITIONEDBY(days(create_time),bucket(user_id,100))文件格式选择-- Parquet推荐列式存储查询性能好TBLPROPERTIES(write.format.defaultparquet)-- ORCHive 生态兼容好TBLPROPERTIES(write.format.defaultorc)-- Avro行式存储适合流式写入TBLPROPERTIES(write.format.defaultavro)压缩配置-- Snappy推荐平衡压缩比和解压速度TBLPROPERTIES(write.parquet.compression-codecsnappy)-- Zstd高压缩比TBLPROPERTIES(write.parquet.compression-codeczstd)-- Gzip最高压缩比但解压慢TBLPROPERTIES(write.parquet.compression-codecgzip)4.2 性能优化小文件合并-- 手动 CompactionCALLprod.system.rewrite_data_files(tableiceberg_db.order_detail,optionsmap(max-concurrent-file-group-rewrites,10));-- 定期 CompactionSpark 调度spark.sql( CALL prod.system.rewrite_data_files( table iceberg_db.order_detail, strategy binpack, options map( max-concurrent-file-group-rewrites, 10, max-file-size, 536870912, -- 512MB target-file-size, 268435456 -- 256MB ) ) )过期快照清理-- 清理 7 天前的快照CALLprod.system.expire_snapshots(tableiceberg_db.order_detail,older_thanTIMESTAMP2026-04-02 00:00:00);-- 保留最近 N 个快照CALLprod.system.expire_snapshots(tableiceberg_db.order_detail,retain_last10);元数据清理-- 清理过期元数据文件CALLprod.system.remove_orphan_files(tableiceberg_db.order_detail,older_thanINTERVAL7DAY);4.3 监控告警关键指标# iceberg_metrics.pyfrompyspark.sqlimportSparkSession sparkSparkSession.builder.appName(Iceberg Metrics).getOrCreate()defget_table_metrics(catalog,database,table):获取表的关键指标# 快照数量snapshot_countspark.sql(fSELECT COUNT(*) FROM{catalog}.{database}.{table}$snapshots).first()[0]# 文件大小分布file_statsspark.sql(fSELECT f COUNT(*) AS file_count, f SUM(file_size_in_bytes) AS total_size, f AVG(file_size_in_bytes) AS avg_size, f MIN(file_size_in_bytes) AS min_size, f MAX(file_size_in_bytes) AS max_size fFROM{catalog}.{database}.{table}.files).first()# 小文件数量 10MBsmall_filesspark.sql(fSELECT COUNT(*) FROM{catalog}.{database}.{table}.files fWHERE file_size_in_bytes 10485760).first()[0]return{snapshot_count:snapshot_count,file_count:file_stats[0],total_size_gb:file_stats[1]/1024/1024/1024,avg_file_size_mb:file_stats[2]/1024/1024,small_files:small_files}# 使用metricsget_table_metrics(prod,iceberg_db,order_detail)print(f快照数{metrics[snapshot_count]})print(f文件数{metrics[file_count]})print(f总大小{metrics[total_size_gb]:.2f}GB)print(f平均文件大小{metrics[avg_file_size_mb]:.2f}MB)print(f小文件数{metrics[small_files]})# 告警ifmetrics[small_files]100:send_alert(fIceberg 表小文件过多{metrics[small_files]})五、生产环境落地案例5.1 案例背景公司某电商平台规模日订单 200 万 日增数据 500GB团队数据团队 35 人建设前痛点Hive 表不支持更新GDPR 合规困难小文件问题严重日均 10 万 小文件数据回滚需要手动恢复备份模式变更需要重写全表5.2 建设方案阶段一离线数仓迁移2 个月- 核心表迁移到 Iceberg50 张表 - Spark 作业改造 - 数据验证和对账阶段二实时数仓建设2 个月- Flink Iceberg 实时写入 - Merge-on-Read 模式 - 自动 Compaction阶段三治理优化持续- 快照过期策略 - 小文件自动合并 - 监控告警体系5.3 建设效果指标建设前建设后提升小文件数量10 万/天 1000/天99% ↓数据回滚时间4-8 小时5 分钟99% ↓GDPR 删除不可行秒级-模式变更重写全表元数据操作99% ↓查询性能基准20%20% ↑六、总结核心要点Iceberg 是 Hive 表的升级- 兼容 Hive Metastore平滑迁移ACID 是核心价值- 解决数据可靠性问题模式演进是生产力- 无需重写数据即可变更 Schema时间旅行是保险- 随时回滚到任意版本多引擎是生态- Spark、Flink、Trino 统一访问最佳实践表设计:-按时间分区天/小时-避免过度分区-Parquet Snappy 格式写入优化:-批量写入用 CoW-流式写入用 MoR-定期 Compaction运维管理:-快照保留 7 天-定期清理 orphan files-监控小文件数量查询优化:-利用谓词下推-避免全表扫描-合理使用时间旅行附录A. 版本兼容性Iceberg 版本SparkFlinkHiveTrino1.0.x3.0-3.21.13-1.142.x3701.1.x3.0-3.31.14-1.152.x3801.2.x3.1-3.31.15-1.163.x3901.4.x3.3-3.41.16-1.173.x400B. 推荐阅读Iceberg 官方文档https://iceberg.apache.org/《Iceberg 权威指南》O’Reilly 2023Iceberg GitHubhttps://github.com/apache/iceberg下一篇《StarRocks/Doris 深度实践》上一篇《数据安全与权限体系》系列目录新技术实战系列