一、前言：为什么需要 Iceberg？

1.1 Hive 的历史包袱

1.2 Netflix 的挑战

1.3 Iceberg 发展历程

二、Iceberg 核心设计

2.1 表格式 vs 存储格式

表格式的职责：

2.2 Iceberg 架构全景

2.3 元数据三层架构

三、元数据详解

3.1 文件布局

iceberg_table/
├── metadata/
│   ├── v1.metadata.json          # 版本1元数据
│   ├── v2.metadata.json          # 版本2元数据
│   ├── v3.metadata.json          # 当前版本
│   ├── snap-1234567890.avro      # 快照文件
│   ├── snap-1234567891.avro
│   └── manifest-list-xxxx.avro   # 清单列表
│   └── manifest-xxxx.avro        # 清单文件
└── data/
    ├── date=2024-01-15/
    │   ├── 00000-0-xxx.parquet
    │   └── 00001-0-xxx.parquet
    └── date=2024-01-16/
        └── 00000-0-xxx.parquet

3.2 Metadata File 结构

{
  "format-version": 2,
  "table-uuid": "xxx-xxx-xxx",
  "location": "s3://bucket/iceberg_table",
  "last-sequence-number": 5,
  "last-updated-ms": 1705312800000,
  "last-column-id": 10,
  "current-schema-id": 1,
  "schemas": [
    {
      "schema-id": 1,
      "fields": [
        {"id": 1, "name": "order_id", "type": "long", "required": true},
        {"id": 2, "name": "user_id", "type": "long"},
        {"id": 3, "name": "amount", "type": "decimal(10,2)"},
        {"id": 4, "name": "order_date", "type": "date"}
      ]
    }
  ],
  "partition-specs": [
    {
      "spec-id": 0,
      "fields": [
        {"source-id": 4, "field-id": 1000, "name": "order_date", "transform": "day"}
      ]
    }
  ],
  "current-snapshot-id": 1234567891,
  "snapshots": [
    {"snapshot-id": 1234567890, "timestamp-ms": 1705312700000, "manifest-list": "..."},
    {"snapshot-id": 1234567891, "timestamp-ms": 1705312800000, "manifest-list": "..."}
  ]
}

3.3 快照与继承

快照继承的好处：

• 只有变化的文件需要记录
• 大量元数据可以复用
• 时间旅行成本低

四、分区进化：Iceberg 的杀手锏

4.1 Hidden Partitioning（隐藏分区）

配置示例：

-- 创建表时定义分区转换
CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    order_time TIMESTAMP
) USING iceberg
PARTITIONED BY (
    days(order_time),    -- 按天分区
    bucket(16, user_id)  -- user_id 分桶
);

-- 查询时无需关心分区
SELECT * FROM orders 
WHERE order_time >= '2024-01-15 00:00:00'
  AND order_time < '2024-01-16 00:00:00';
-- Iceberg 自动进行分区裁剪！

4.2 分区转换函数

4.3 分区演进（Partition Evolution）

-- 分区演进：从月分区改为天分区
ALTER TABLE orders 
SET PARTITION SPEC (
    day(order_time),
    bucket(16, user_id)
);

-- 旧数据仍然是月分区，新数据是天分区
-- 查询时 Iceberg 自动处理！

五、Schema 演进

5.1 支持的演进操作

5.2 Schema 演进原理

示例：

-- 添加列
ALTER TABLE orders ADD COLUMN discount DECIMAL(10,2);

-- 重命名列
ALTER TABLE orders RENAME COLUMN discount TO discount_amount;

-- 删除列
ALTER TABLE orders DROP COLUMN discount_amount;

-- 类型变更（安全变更）
ALTER TABLE orders ALTER COLUMN user_id TYPE BIGINT;

5.3 类型演进规则

六、时间旅行

6.1 基于快照的时间旅行

6.2 时间旅行查询

-- 方式1：按快照ID查询
SELECT * FROM orders VERSION AS OF 1234567890;

-- 方式2：按时间戳查询
SELECT * FROM orders TIMESTAMP AS OF '2024-01-15 10:00:00';

-- 方式3：按时间表达式
SELECT * FROM orders TIMESTAMP AS OF current_timestamp() - INTERVAL 1 HOUR;

-- Spark SQL 语法
SELECT * FROM orders.snapshots;  -- 查看所有快照
SELECT * FROM orders.history;    -- 查看表历史

6.3 回滚操作

-- 回滚到指定快照
CALL system.rollback_to_snapshot('db.orders', 1234567890);

-- 回滚到指定时间点
CALL system.rollback_to_timestamp('db.orders', TIMESTAMP '2024-01-15 10:00:00');

七、ACID 事务与并发控制

7.1 乐观并发控制

7.2 行级更新与删除

Delete File 类型：

-- 行级删除
DELETE FROM orders WHERE order_id = 12345;

-- 行级更新
UPDATE orders SET status = 'completed' WHERE order_id = 12345;

-- Merge 操作
MERGE INTO orders t
USING updates s ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET t.status = s.status
WHEN NOT MATCHED THEN INSERT *;

八、查询优化

8.1 元数据过滤

8.2 列级统计信息

8.3 Data Skipping 示例

-- 假设有1000个数据文件
SELECT * FROM orders 
WHERE order_date = '2024-01-15' 
  AND amount > 1000;

-- Iceberg 优化流程：
-- 1. 通过 Manifest List 的分区统计，定位到 order_date='2024-01-15' 的 Manifest
-- 2. 通过 Manifest File 的列统计，跳过 max(amount) < 1000 的文件
-- 3. 最终可能只需要读取 10 个文件

九、实战操作

9.1 创建 Iceberg 表

-- Spark SQL 创建表
CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    status STRING,
    order_time TIMESTAMP
) USING iceberg
PARTITIONED BY (days(order_time))
LOCATION 's3://bucket/iceberg/orders'
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'zstd',
    'write.metadata.delete-after-commit.enabled' = 'true',
    'write.metadata.previous-versions-max' = '100'
);

9.2 数据写入

// Spark DataFrame 写入
val df = spark.createDataFrame(Seq(
  (1L, 1001L, BigDecimal("100.00"), "pending", Timestamp.valueOf("2024-01-15 10:00:00")),
  (2L, 1002L, BigDecimal("200.00"), "completed", Timestamp.valueOf("2024-01-15 11:00:00"))
)).toDF("order_id", "user_id", "amount", "status", "order_time")

// 追加写入
df.writeTo("orders").append()

// 覆盖写入
df.writeTo("orders").overwritePartitions()

9.3 Flink 流式写入

-- Flink SQL 创建 Iceberg Sink
CREATE TABLE iceberg_orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10,2),
    status STRING,
    order_time TIMESTAMP(3),
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'iceberg',
    'catalog-type' = 'hive',
    'catalog-name' = 'hive_catalog',
    'warehouse' = 's3://bucket/iceberg',
    'format-version' = '2',
    'write.upsert.enabled' = 'true'
);

-- 流式 Upsert
INSERT INTO iceberg_orders
SELECT * FROM kafka_orders;

9.4 表维护操作

-- 查看快照
SELECT * FROM orders.snapshots;

-- 查看文件
SELECT * FROM orders.files;

-- 查看历史
SELECT * FROM orders.history;

-- 过期快照清理
CALL system.expire_snapshots('db.orders', TIMESTAMP '2024-01-01 00:00:00', 100);

-- 删除孤儿文件
CALL system.remove_orphan_files('db.orders');

-- 重写小文件
CALL system.rewrite_data_files('db.orders');

-- 重写 Manifest
CALL system.rewrite_manifests('db.orders');

十、与引擎集成

10.1 多引擎支持

10.2 Catalog 选择

十一、最佳实践

11.1 配置建议

# 写入配置
write.format.default=parquet
write.parquet.compression-codec=zstd
write.target-file-size-bytes=536870912  # 512MB

# 分区配置
write.distribution-mode=hash  # 写入时hash分布

# 元数据配置
write.metadata.delete-after-commit.enabled=true
write.metadata.previous-versions-max=100

# 读取配置
read.split.target-size=134217728  # 128MB

11.2 性能优化

十二、总结

12.1 Iceberg 核心优势

12.2 适用场景

12.3 一句话总结

Iceberg = 开放表格式 + 隐藏分区 + 无缝 Schema 演进 + 时间旅行

如果你需要一个开放、灵活、多引擎支持的数据湖表格式，Iceberg 是当前最佳选择之一。