Apache Iceberg 最佳实践与高级特性

CREATE TABLE my_table (...)
USING iceberg
TBLPROPERTIES (
  -- 格式与版本
  'format' = 'iceberg/parquet',
  'format-version' = '2',
  
  -- 事务与更新
  'write.upsert.enabled' = 'true',
  'write.merge.mode' = 'merge-on-read',
  'identifier-fields' = '[user_id, event_date]',
  
  -- 历史保留
  'history.expire.max-snapshot-age-ms' = '259200000',  -- 3天
  'history.expire.max-ref-age-ms' = '259200000'
)

-- 按日期分区的示例
CREATE TABLE events (
  event_time timestamp,
  user_id bigint,
  event_type string,
  data map<string, string>
)
USING iceberg
PARTITIONED BY (days(event_time))

顶层(表历史) → 中层(快照和清单) → 底层(数据文件)

history ──┐
          ├→ snapshots ──┐
refs ─────┘              ├→ manifests ─→ files
metadata_log_entries ────┘              │
                       partitions ──────┘

-- 找出特定快照中新增的大文件
SELECT f.file_path, f.file_size_in_bytes, f.record_count
FROM my_table.files f
JOIN my_table.manifests m ON f.manifest_path = m.path
WHERE m.added_snapshot_id = 1234567890
AND f.file_size_in_bytes > 100000000
ORDER BY f.file_size_in_bytes DESC;

-- 跟踪特定分区的变更历史
SELECT s.committed_at, s.operation, p.record_count
FROM my_table.snapshots s
JOIN my_table.partitions p
  ON p.partition = 'date=2025-05-25'
WHERE s.snapshot_id >= 
  (SELECT min(snapshot_id) FROM my_table.manifests 
   WHERE path IN (SELECT manifest_path FROM my_table.files 
                  WHERE partition = 'date=2025-05-25'))
ORDER BY s.committed_at;

-- 查看表的历史变更
SELECT * FROM my_table.history ORDER BY made_current_at DESC;

-- 查看表的快照
SELECT snapshot_id, operation, committed_at, summary 
FROM my_table.snapshots 
ORDER BY committed_at DESC;

-- 查看表的数据文件
SELECT file_path, record_count, file_size_in_bytes 
FROM my_table.files 
ORDER BY file_size_in_bytes DESC;

-- 查看表的清单文件
SELECT path, added_snapshot_id, added_data_files_count 
FROM my_table.manifests;

-- 查看表的分区信息
SELECT partition, record_count, file_count 
FROM my_table.partitions 
ORDER BY record_count DESC;

-- 查看表的引用
SELECT * FROM my_table.refs;

-- 查找仅包含最大快照规则的引用
SELECT name, min_snapshots_to_keep, max_snapshot_age_in_ms
FROM my_table.refs
WHERE min_snapshots_to_keep IS NOT NULL AND max_snapshot_age_in_ms IS NOT NULL;

-- 将表回滚到特定快照
CALL catalog.system.set_current_snapshot('my_table', 1234567890);

-- 将main分支更新到feature分支的最新状态
CALL catalog.system.fast_forward('my_table', 'main', 'feature-branch');

-- 将特定快照的变更应用到main分支
CALL catalog.system.cherrypick_snapshot(
  'my_table', 
  'main', 
  snapshot_id,
  map('cherry-pick.conflict-resolution-mode', 'PRESERVE_MAIN')
);

-- 创建新分支
CALL catalog.system.create_branch('my_table', 'feature-x', snapshot_id);

-- 在特定分支上操作
INSERT INTO my_table FOR BRANCH 'feature-x' VALUES (...);

-- 启用WAP
ALTER TABLE my_table SET TBLPROPERTIES (
  'write.wap.enabled' = 'true'
);

-- 使用WAP (Spark示例)
spark.conf.set("spark.wap.id", "wap-job-123");
spark.sql("INSERT INTO my_table SELECT * FROM source_table");

-- WAP高级配置
ALTER TABLE my_table SET TBLPROPERTIES (
  'write.wap.enabled' = 'true',
  'write.object-storage.enabled' = 'true',  -- 对象存储优化
  'write.metadata.delete-after-commit.enabled' = 'true',  -- 延迟删除
  'write.folder-storage.path-style' = 'wap-friendly'  -- 路径优化
);

-- Spark SQL中的自动恢复配置
SET spark.sql.iceberg.handle-wap-conflicts = 'auto';
SET spark.sql.iceberg.wap.auto-resume-enabled = 'true';
SET spark.sql.iceberg.wap.conflict-resolution-mode = 'preserve-incoming';

-- 使用有意义的WAP ID
SET spark.wap.id = 'daily-etl-2025-05-27';

-- 启用回顾窗口优化
ALTER TABLE my_table SET TBLPROPERTIES (
  'read.split.planning-lookback.enabled' = 'true',
  'read.split.planning-lookback.window' = '2592000000'  -- 30天（毫秒）
);

-- 设置清单文件优化参数
ALTER TABLE my_table SET TBLPROPERTIES (
  'commit.manifest.min-count-to-merge' = '100',  -- 低于此值的清单将被合并
  'commit.manifest.target-size-bytes' = '8388608',  -- 目标大小8MB
  'commit.manifest.max-count' = '5000'  -- 单个清单最多5000个条目
);

-- 主动重写清单文件
CALL catalog.system.rewrite_manifests(
  table => 'my_table',
  strategy => 'binpack'
);

-- 配置文件大小
ALTER TABLE my_table SET TBLPROPERTIES (
  'write.target-file-size-bytes' = '536870912',  -- 512MB
  'write.distribution-mode' = 'hash'
);

-- 运行数据文件压缩
CALL catalog.system.rewrite_data_files(
  table => 'my_table',
  strategy => 'binpack',
  options => map('min-input-files','5', 'target-file-size-bytes', '536870912')
);

快照过期 → 垃圾收集 → 物理文件删除

-- 表级过期配置（安全网）
ALTER TABLE my_table SET TBLPROPERTIES (
  'history.expire.min-snapshots-to-keep' = '20',  -- 保留至少20个快照
  'history.expire.max-snapshot-age-ms' = '2592000000'  -- 30天
);

-- 主动运行维护作业（更积极清理）
CALL catalog.system.expire_snapshots(
  table => 'my_table',
  older_than => CURRENT_TIMESTAMP() - INTERVAL 7 DAY,  -- 保留7天
  retain_last => 10  -- 至少保留10个快照
);

-- 查找过期快照的ID
SELECT snapshot_id, committed_at 
FROM my_table.snapshots
WHERE committed_at > TIMESTAMP '2025-05-01 00:00:00'
ORDER BY committed_at;

-- 使用特定快照创建新分支
CALL catalog.system.create_branch(
  table => 'my_table',
  branch => 'recovered_snapshot',
  snapshot_id => 123456789
);

ALTER TABLE standard_table SET TBLPROPERTIES (
  -- 快照过期配置
  'history.expire.min-snapshots-to-keep' = '20',
  'history.expire.max-snapshot-age-ms' = '1209600000',  -- 14天
  
  -- 垃圾收集配置
  'gc.enabled' = 'true',
  'gc.min-snapshots-to-keep' = '5',  -- 额外保护5个快照
  
  -- 元数据优化
  'write.metadata.delete-after-commit.enabled' = 'true',
  
  -- 孤儿文件清理配置
  'write.orphan-files.purge.enabled' = 'true',
  'write.orphan-files.retention-duration' = '1209600000'  -- 14天
);

ALTER TABLE high_frequency_table SET TBLPROPERTIES (
  -- 快照过期较短
  'history.expire.min-snapshots-to-keep' = '10',
  'history.expire.max-snapshot-age-ms' = '259200000',  -- 3天
  
  -- 积极的垃圾收集
  'gc.enabled' = 'true',
  'gc.min-snapshots-to-keep' = '3',
  
  -- 更频繁的孤儿文件清理
  'write.orphan-files.purge.enabled' = 'true',
  'write.orphan-files.retention-duration' = '259200000'  -- 3天
);

-- 启用延迟元数据删除
ALTER TABLE my_table SET TBLPROPERTIES (
  'write.metadata.delete-after-commit.enabled' = 'true',
  'write.metadata.previous-versions-max' = '50'  -- 保留50个版本
);

-- 添加新列
ALTER TABLE my_table ADD COLUMN user_email string AFTER user_id;

-- 重命名列
ALTER TABLE my_table RENAME COLUMN user_id TO customer_id;

-- 设置名称映射
ALTER TABLE my_table SET TBLPROPERTIES (
  'schema.name-mapping.default' = '{
    "mappings": [
      {"field-id": 1, "names": ["item_id", "item_identifier"]},
      {"field-id": 2, "names": ["transaction_id", "txn_id"]}
    ]
  }'
);

-- 创建变更日志视图
CALL catalog.system.create_changelog_view( 
  table => 'my_db.my_table',
  options => map(
    'start-snapshot-id', '4816648710583642722',
    'end-snapshot-id', '2557325773776943708'
  )
)

-- 设置WAP ID
SET spark.wap.id = 'test-branch-wap-123';

-- 在分支上执行多个操作
INSERT INTO my_table FOR BRANCH 'test-branch' SELECT * FROM source_1;
UPDATE my_table FOR BRANCH 'test-branch' SET col1 = 'new_value' WHERE id = 5;
DELETE FROM my_table FOR BRANCH 'test-branch' WHERE status = 'expired';

-- 提交或回滚整个事务
CALL catalog.system.commit_transaction('test-branch-wap-123');
-- 或
CALL catalog.system.rollback_transaction('test-branch-wap-123');

-- 使用同一个WAP ID处理多个表
SET spark.wap.id = 'multi-table-transaction-123';

-- 对第一个表的操作
INSERT INTO table1 FOR BRANCH 'test-branch' VALUES (...);

-- 对第二个表的操作
UPDATE table2 FOR BRANCH 'test-branch' SET col1 = 'new_value' WHERE id = 5;

-- 提交整个事务（影响两个表）
CALL catalog.system.commit_transaction('multi-table-transaction-123');

-- 测试不同的冲突解决模式
CALL catalog.system.cherrypick_snapshot(
  'my_table', 
  'main-branch', 
  snapshot_id,
  map('cherry-pick.conflict-resolution-mode', 'PRESERVE_SOURCE')
);

-- 按版本号查询历史数据
SELECT * FROM my_table VERSION AS OF 5347521240857739485;

-- 按时间点查询历史数据
SELECT * FROM my_table FOR TIMESTAMP AS OF '2025-05-20 10:00:00';

-- 获取最近24小时的变更
SELECT * FROM my_table AS OF TIMESTAMP current_timestamp()
EXCEPT 
SELECT * FROM my_table AS OF TIMESTAMP current_timestamp() - interval 1 day;

-- 监控小文件问题
SELECT count(*) as small_files_count 
FROM my_table.files 
WHERE file_size_in_bytes < 1024*1024;

-- 识别需要压缩的分区
SELECT partition, file_count 
FROM my_table.partitions 
WHERE file_count > 50
ORDER BY file_count DESC;

初始状态：
- 数据文件: data-file-001.parquet (包含100条记录)
- Manifest: manifest-001.avro (引用 data-file-001)
- 快照: snapshot-001 (引用 manifest-001)

执行更新操作 (Merge-on-Read 模式):
UPDATE my_table SET col1 = 'new_value' WHERE id = 5;

更新后状态:
- 数据文件: 
  * data-file-001.parquet (不变，保留原始数据)
  * delete-file-002.parquet (标识 id=5 的记录被删除)
  * data-file-003.parquet (包含更新后的 id=5 记录)
- Manifest:
  * manifest-001.avro (不变，但在新快照中不再使用)
  * manifest-002.avro (引用所有三个数据文件，并标记它们的状态)
- 快照: 
  * snapshot-001 (不变，但不再是当前快照)
  * snapshot-002 (新的当前快照，引用 manifest-002)