数据存储 Iceberg

第1章：Iceberg 概述与核心理念

1.1 什么是 Iceberg？

概念名称	说明	注意事项
Apache Iceberg	一种开放的表格式（Table Format），用于大规模分析型数据湖，提供高性能的元数据管理与事务支持。	不是存储引擎，也不提供存储，而是定义如何组织数据文件与元数据。
表格式（Table Format）	定义数据表的元数据结构、文件组织方式、事务语义等，位于文件格式（如 Parquet、ORC）之上。	区别于传统 Hive 表的 metastore，Iceberg 使用自描述的元数据文件。
开放标准	Iceberg 格式完全开放，支持多引擎（Spark、Flink、Trino 等）互操作。	所有元数据以开放格式（如 Avro、JSON）存储，便于解析与审计。
高性能元数据	通过分层元数据设计（Metadata → Manifest → Data Files）避免全表扫描。	支持谓词下推到文件级别，提升查询效率。

1.2 Iceberg 的诞生背景与解决的问题

问题背景	说明	注意事项
Hive 表的局限性	Hive 表依赖 Hive Metastore 存储 schema 与分区信息，无法支持行级更新、时间旅行等高级功能。	元数据操作（如分区重命名）非原子性，易出错。
小文件问题严重	大量小文件导致 NameNode 压力大，查询性能下降。	Iceberg 提供 rewrite_data_files 等操作优化文件布局。
缺乏事务支持	传统数据湖无法保证写入的原子性与隔离性。	Iceberg 提供快照级事务，支持 ACID。
数据一致性差	多个写入任务可能导致数据重复或丢失。	Iceberg 使用乐观锁机制保证并发写入一致性。
时间旅行缺失	无法回溯历史版本数据用于审计或恢复。	Iceberg 快照机制天然支持 AS OF 查询。

1.3 Iceberg 与其他数据湖格式（Hudi、Delta Lake）对比

对比维度	Iceberg	Hudi	Delta Lake	注意事项
开发公司	Netflix（现 Apache 项目）	Uber	Databricks	Delta Lake 与 Databricks 深度绑定。
核心定位	表格式（Table Format）	增量处理框架 + 表格式	表格式 + 引擎扩展	Iceberg 更中立，支持多引擎原生集成。
更新粒度	支持 MERGE INTO（行级）	支持 upsert（行级）	支持 upsert（行级）	Hudi 和 Delta Lake 更擅长实时更新。
架构设计	分层元数据（Metadata → Manifest → Data）	Log + Snapshot（COW/MOR）	Transaction Log + Parquet	Iceberg 元数据更轻量，适合超大规模表。
时间旅行	支持（基于快照）	支持（基于 commits）	支持（基于 version）	三者均支持，语法略有差异。
分支与标签	支持（0.13+）	不支持	支持（Unity Catalog）	Iceberg 分支可用于测试、回滚等场景。
小文件合并	支持 rewrite_data_files	支持 compaction	支持 OPTIMIZE	Iceberg 提供更灵活的重写策略。
流式写入支持	支持（Streaming Write）	原生支持（MOR）	原生支持（StreamWriter）	Hudi 和 Delta Lake 在流式场景更成熟。

1.4 Iceberg 的核心设计哲学

设计理念	说明	注意事项
表格式抽象	将”表”定义为一组元数据文件 + 数据文件，独立于计算引擎。	任何支持 Iceberg 协议的引擎均可读写同一张表。
元数据层级化	使用 Table Metadata → Manifest List → Manifest File → Data File 四层结构。	避免查询时扫描所有数据文件，提升性能。
快照隔离（Snapshot Isolation）	每次提交生成一个新快照，读写操作基于快照进行，保证一致性。	读操作不阻塞写操作，支持高并发。
原子性提交	使用”原子替换”元数据文件方式实现事务提交，要么全部成功，要么失败。	依赖底层文件系统支持原子 rename（如 HDFS、S3A）。
隐藏分区（Hidden Partitioning）	分区逻辑由 Iceberg 管理，用户写入无需指定分区路径。	减少人为错误，支持分区演化。
可扩展性	支持自定义分区策略、文件格式、加密、压缩等。	通过 Table Properties 配置，易于扩展。

第2章：Iceberg 架构与存储模型

2.1 表结构：Metadata File、Manifest File、Data File 层级解析

文件类型	说明	注意事项
Data File	存储实际数据的文件，格式通常为 Parquet、ORC 或 Avro。	文件路径由 Iceberg 自动生成，遵循分区布局。
Manifest File	记录一组 Data File 的元信息（路径、大小、行数、统计信息等）。	支持谓词下推（如 min/max 值），跳过不相关文件。
Manifest List	又称”快照文件”，记录一个快照包含的所有 Manifest Files。	每个快照对应一个 Manifest List，文件名为 snap-*.avro。
Metadata File	表的元数据文件，包含 schema、分区策略、当前快照 ID、历史快照列表等。	文件以 metadata/<uuid>.metadata.json 存储，使用快照指针引用。
Metadata Log	记录所有 Metadata File 的变更历史（保留最近 N 个）。	用于恢复和审计，由 write.metadata.previous-versions-max 控制数量。

2.2 快照（Snapshot）机制与时间旅行原理

概念	说明	注意事项
快照（Snapshot）	表在某一时刻的数据视图，由一个唯一的 snapshot-id 标识。	每次写入（insert/update/delete）都会生成新快照。
快照链（Snapshot Chain）	所有快照按时间顺序形成链式结构，支持回溯。	可通过 table.history() 查看历史快照。
时间旅行（Time Travel）	查询历史快照数据，语法如 SELECT * FROM table TIMESTAMP AS OF '2024-01-01 00:00:00'。	快照未过期前提下可用，受 expire_snapshots 策略影响。
快照隔离读	读操作基于某个快照进行，不会看到未提交的写入。	实现 MVCC（多版本并发控制），提升并发性能。
快照元数据存储	每个快照在 metadata/ 目录下生成一个 Manifest List 文件。	快照文件不包含数据，仅记录文件列表。

2.3 分支（Branch）与标签（Tag）支持（自 0.13+）

概念	说明	注意事项
主分支（main）	默认分支，所有写入默认提交至此。	类似 Git 的 main 分支。
自定义分支（Branch）	可创建独立分支用于测试、隔离写入。	语法：CALL branch('table', 'dev')。
标签（Tag）	指向某个快照的静态引用，常用于版本标记。	语法：CALL tag('table', 'v1.0')。
分支写入	写入可指定分支，不影响主分支。	需在写入时设置 write.branch 属性。
分支查询	查询时可指定分支。	语法：SELECT * FROM table FOR VERSION AS OF branch='dev'，分支支持时间旅行与并发开发。

2.4 元数据层级：Table Metadata 与 Snapshot Chain

层级	说明	注意事项
Table Metadata（最新）	当前表的元数据，指向当前快照（current-snapshot-id）。	文件位于 metadata/<uuid>.metadata.json。
Table Metadata（历史）	过去版本的元数据文件，保留于 metadata/ 目录。	用于恢复和审计，数量受 previous-versions-max 限制。
Snapshot Chain	由 current-snapshot-id 指向当前快照，快照中包含 parent-snapshot-id 形成链。	支持回溯到任意历史状态。
Metadata Log	记录所有元数据文件的变更，存储于 metadata/metadata-log/<timestamp>.txt。	每行记录一个元数据文件的路径与时间戳。
元数据一致性	通过原子写入和文件系统 rename 保证一致性。	不支持原子性的存储（如 S3）需启用 s3a.committer。

2.5 文件组织方式：Partitioning Strategies（分区策略）

分区策略	语法示例	用途	代码示例	注意事项
Identity	identity(day)	直接使用列值作为分区名。	PARTITIONED BY (day)	适用于低基数列。
Year	year(ts)	按时间戳的年份分区。	PARTITIONED BY (year(ts))	分区名格式：year=2024。
Month	month(ts)	按月份分区。	PARTITIONED BY (month(ts))	支持时间类型（timestamp、date）。
Day	day(ts)	按天分区。	PARTITIONED BY (day(ts))	最常用的时间分区策略。
Hour	hour(ts)	按小时分区。	PARTITIONED BY (hour(ts))	适用于高频率写入场景。
Bucket	bucket(id, 16)	按哈希桶分区，共 16 个桶。	PARTITIONED BY (bucket(id, 16))	避免倾斜，提升 Join 性能。
Truncate	truncate(name, 10)	按前 10 个字符截断分区。	PARTITIONED BY (truncate(name, 10))	适用于字符串列。
分区演化	ALTER TABLE t ADD PARTITION FIELD year(ts)	动态添加分区策略。	ALTER TABLE logs ADD PARTITION FIELD hour(ts)	支持向后兼容，旧数据不重写。

第3章：环境准备与快速入门

3.1 支持的计算引擎概述（Spark、Flink、Trino、Presto、Hive 等）

计算引擎	支持程度	说明	注意事项
Apache Spark	完全支持（官方首选）	提供 Spark Datasource 和 Spark SQL 集成，支持读写、分区、时间旅行等全部功能。	推荐使用 Spark 3.2+ 与 Iceberg 0.14+ 版本。
Apache Flink	完全支持（流批一体）	支持 Streaming Write 和 Batch Read，适用于实时数仓场景。	需引入 iceberg-flink-runtime 依赖。
Trino (formerly PrestoSQL)	完全支持	支持高效查询 Iceberg 表，兼容 Hive Metastore 或 Glue Catalog。	查询性能优异，支持谓词下推与分区裁剪。
PrestoDB (Facebook)	支持（社区版）	基本查询功能可用，但功能不如 Trino 完善。	建议优先使用 Trino。
Apache Hive	有限支持	通过 IcebergStorageHandler 支持读写，但不支持所有高级特性（如 MERGE INTO）。	仅推荐用于只读查询或迁移场景。
Impala	实验性支持	Cloudera 提供有限支持，功能不完整。	不建议在生产环境使用。
BigQuery / Redshift	不支持原生读取	需通过外部表或导出方式集成。	可借助 Glue Catalog + Trino 作为中间层。

3.2 基于 Spark 的 Iceberg 环境搭建（Local/Cluster）

方法/配置项	语法	用途	代码示例	注意事项
启动 Spark with Iceberg JAR	--packages org.apache.iceberg:iceberg-spark3-runtime:0.14.1	下载并加载 Iceberg 运行时依赖	spark-shell --packages org.apache.iceberg:iceberg-spark3-runtime:0.14.1 --conf spark.sql.catalog.demo.type=rest --conf spark.sql.catalog.demo.uri=https://api.example.com --conf spark.sql.catalog.demo.ref=main	推荐使用 --packages 自动解析依赖。
配置 Catalog 类型	spark.sql.catalog.<name>.type	指定 Catalog 实现类型	--conf spark.sql.catalog.local=iceberg --conf spark.sql.catalog.local.type=hive --conf spark.sql.catalog.local.uri=thrift://hive-metastore:9083	支持 hive、hadoop、rest、glue。
指定默认 Catalog	spark.sql.catalogImplementation	设置默认表注册中心	--conf spark.sql.catalogImplementation=iceberg	启用后所有非限定表名默认使用 Iceberg。
Hadoop Catalog 配置	spark.sql.catalog.<name>.warehouse	指定表根路径	--conf spark.sql.catalog.local=iceberg --conf spark.sql.catalog.local.type=hadoop --conf spark.sql.catalog.local.warehouse=s3a://my-bucket/warehouse	适用于无 Hive Metastore 场景。
REST Catalog 配置	多项配置组合	使用 Iceberg REST Catalog 服务	--conf spark.sql.catalog.rest.type=rest --conf spark.sql.catalog.rest.uri=https://iceberg-api.example.com --conf spark.sql.catalog.rest.ref=main --conf spark.sql.catalog.rest.credentials=token:xxx	适合多引擎共享元数据。
S3 访问配置	spark.hadoop.fs.s3a.access.key 等	配置对象存储访问凭证	--conf spark.hadoop.fs.s3a.access.key=AKIA... --conf spark.hadoop.fs.s3a.secret.key=xxxx --conf spark.hadoop.fs.s3a.endpoint=s3.amazonaws.com	若使用 S3，必须正确配置权限。

3.3 创建第一个 Iceberg 表（Spark SQL）

方法	语法	用途	代码示例	注意事项
CREATE TABLE	CREATE TABLE [catalog.]db.table (col type, ...) USING iceberg	创建新 Iceberg 表	CREATE TABLE local.db.sample (id BIGINT, name STRING, dt DATE) USING iceberg;	必须指定列和类型。
指定 LOCATION	LOCATION 'path'	自定义表存储路径	CREATE TABLE local.db.sample (id BIGINT, name STRING) USING iceberg LOCATION 's3a://bucket/tables/sample';	路径必须为空或不存在。
添加 TABLE PROPERTIES	TBLPROPERTIES ('key'='value')	设置表级配置	CREATE TABLE local.db.sample (id BIGINT) USING iceberg TBLPROPERTIES ('write.format.default'='parquet');	常见属性：format、compression、commit.retry 等。
分区创建	PARTITIONED BY (col)	按列分区	CREATE TABLE logs (user_id BIGINT, ts TIMESTAMP) USING iceberg PARTITIONED BY (day(ts));	支持隐藏分区函数。
指定 Catalog 和 Database	catalog.db.table	显式指定命名空间	CREATE TABLE local.production.users (id BIGINT, email STRING) USING iceberg;	需提前创建 database。

3.4 插入与查询数据（基础 CRUD 操作演示）

操作	语法	用途	代码示例	注意事项
插入数据（INSERT INTO）	INSERT INTO table SELECT ...	追加写入数据	INSERT INTO local.db.sample SELECT id, name FROM source_table;	不覆盖已有数据。
覆盖写入（INSERT OVERWRITE）	INSERT OVERWRITE table ...	覆盖全表或分区数据	INSERT OVERWRITE local.db.sample SELECT id, name FROM cleaned_source;	支持分区覆盖（WHERE 条件）。
查询数据（SELECT）	SELECT * FROM table	读取表内容	SELECT * FROM local.db.sample WHERE id > 100;	支持标准 SQL 查询。
时间旅行查询	FOR SYSTEM_TIME AS OF	查询历史版本	SELECT * FROM local.db.sample FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00';	快照未过期前提下可用。
查询快照历史	table.history	查看快照链	SELECT * FROM local.db.sample.history;	返回 snapshot_id、committed_at 等。
查询文件信息	table.files	查看数据文件分布	SELECT * FROM local.db.sample.files;	包含 file_path、file_size、record_count。
删除表中数据	DELETE FROM table WHERE ...	行级删除（需匹配主键或条件）	DELETE FROM local.db.sample WHERE id = 1;	Iceberg 0.12+ 支持。

第4章：Iceberg 表的创建与管理（Spark SQL）

4.1 CREATE TABLE 语句详解（含 LOCATION、PARTITIONED BY、TABLE PROPERTIES）

参数/子句	语法	用途	代码示例	注意事项
TABLE 名称	[catalog.][database.]table_name	指定表的完整路径	CREATE TABLE local.sales.orders (...)	若未指定 catalog，使用当前默认。
列定义	column_name DATA_TYPE [COMMENT '...']	定义字段名与类型	id BIGINT COMMENT '用户ID', name STRING	支持嵌套类型（struct、array、map）。
USING 子句	USING iceberg	指定使用 Iceberg 表格式	USING iceberg	必须显式声明。
LOCATION 子句	LOCATION 'path'	指定表数据存储位置	LOCATION 's3a://mybucket/tables/orders'	路径必须为空或不存在。
PARTITIONED BY	PARTITIONED BY (transform)	定义分区策略	PARTITIONED BY (day(ts), bucket(user_id, 16))	支持复合分区。
TABLE PROPERTIES	TBLPROPERTIES ('key'='value')	设置表级配置	TBLPROPERTIES ('write.format.default'='parquet', 'commit.retry.num'='3')	常见属性见下表。

常用 TABLE PROPERTIES 配置表：

属性名	默认值	说明	注意事项
write.format.default	parquet	指定写入文件格式（parquet/orc/avro）	推荐使用 parquet。
write.parquet.compression-codec	gzip	Parquet 压缩算法（gzip/snappy/zstd）	snappy 性能更好。
write.metadata.previous-versions-max	100	保留历史 metadata 文件数量	控制元数据存储成本。
commit.retry.num	4	提交重试次数	网络不稳定时可调大。
format-version	1	表格式版本（1 或 2）	v2 支持行级删除、位置删除等。

4.2 使用 EXISTING LOCATION 创建外部表

方法	语法	用途	代码示例	注意事项
CREATE TABLE … EXISTING LOCATION	CREATE TABLE ... LOCATION 'path' USING iceberg	关联已有 Iceberg 表路径	CREATE TABLE local.db.existing_table USING iceberg LOCATION 's3a://bucket/tables/existing';	路径必须包含有效的 metadata/ 目录。
验证表结构	DESCRIBE table	查看表 schema	DESCRIBE local.db.existing_table;	确认字段与分区正确。
指定 Catalog	catalog.db.table	在指定 catalog 中创建引用	CREATE TABLE hive_catalog.default.logs USING iceberg LOCATION 's3a://data/logs';	支持跨 catalog 引用。
不支持的操作	-	不能使用 PARTITIONED BY 或 TBLPROPERTIES	-	由原表决定，不能覆盖。
路径要求	-	LOCATION 必须指向 Iceberg 表根目录	-	必须包含 metadata/ 子目录。

4.3 ALTER TABLE 常用操作（重命名、添加/删除/重命名列、修改属性）

操作	语法	用途	代码示例	注意事项
添加列	ALTER TABLE ADD COLUMN	新增字段	ALTER TABLE local.db.sample ADD COLUMN email STRING;	支持嵌套字段（如 ADD COLUMN address.city STRING）。
删除列	ALTER TABLE DROP COLUMN	删除字段	ALTER TABLE local.db.sample DROP COLUMN tmp_col;	需设置 spark.sql.legacy.alterTableDropColumn.enabled=true。
重命名列	ALTER TABLE RENAME COLUMN	修改字段名	ALTER TABLE local.db.sample RENAME COLUMN name TO full_name;	不影响数据文件，仅更新元数据。
修改列类型	ALTER TABLE CHANGE COLUMN	修改字段数据类型	ALTER TABLE local.db.sample CHANGE COLUMN id id BIGINT;	仅支持兼容类型转换（如 int → bigint）。
添加分区字段	ALTER TABLE ADD PARTITION FIELD	动态添加分区策略	ALTER TABLE logs ADD PARTITION FIELD hour(ts);	分区演化，旧数据按新规则查询。
删除分区字段	ALTER TABLE DROP PARTITION FIELD	移除分区策略	ALTER TABLE logs DROP PARTITION FIELD day(ts);	需 Iceberg 0.13+ 支持。
设置表属性	ALTER TABLE SET TBLPROPERTIES	修改表配置	ALTER TABLE local.db.sample SET TBLPROPERTIES ('write.zorder.enabled'='true');	影响后续写入行为。
删除表属性	ALTER TABLE UNSET TBLPROPERTIES	移除属性	ALTER TABLE local.db.sample UNSET TBLPROPERTIES ('temp_flag');	恢复为默认值。

4.4 DROP TABLE 与表生命周期管理

操作	语法	用途	代码示例	注意事项
删除表（保留数据）	DROP TABLE [IF EXISTS] table	仅删除元数据，数据文件保留	DROP TABLE local.db.temp_table;	数据仍可被外部访问。
彻底删除表（数据+元数据）	手动清理或脚本	删除表后清理存储路径	hdfs dfs -rm -r /path/to/table 或 s3 rm s3://bucket/table -r	Iceberg 不自动清理数据文件。
判断表是否存在	DROP TABLE IF EXISTS	避免删除不存在的表报错	DROP TABLE IF EXISTS local.db.tmp;	推荐用于脚本中。
外部表删除	DROP TABLE	仅删除元数据引用	DROP TABLE local.db.external_view;	不影响原数据路径。
生命周期策略	配合外部工具	自动化清理过期表	使用 Airflow 调度 Python 脚本，查询 information_schema.tables，按 create_time 删除 >30 天的表。	建议建立表管理规范。
元数据保留策略	write.metadata.previous-versions-max	控制历史 metadata 文件数量	设置 spark.sql.catalog.local['write.metadata.previous-versions-max']='5'	防止元数据膨胀。

第5章：数据写入与更新操作

5.1 使用 Spark DataFrame 写入 Iceberg 表（append、overwrite、overwritePartitions）

写入模式	语法	用途	代码示例	注意事项
append	.mode("append")	追加数据到表中，不修改已有数据	df.write.mode("append").format("iceberg").saveAsTable("local.db.users")	默认模式，适用于日志、事件数据。
overwrite	.mode("overwrite")	覆盖整张表的数据	df.write.mode("overwrite").format("iceberg").saveAsTable("local.db.users")	生成新快照，旧数据仍可通过时间旅行访问。
overwritePartitions	.mode("overwrite").option("replaceWhere", condition)	仅覆盖满足条件的分区	df.write.mode("overwrite").option("replaceWhere", "dt = '2024-01-01'").format("iceberg").saveAsTable("local.db.users")	replaceWhere 必须与分区字段匹配，否则报错。
指定分区写入	无需额外配置	写入自动按分区组织	df.withColumn("dt", current_date()).write.partitionBy("dt").mode("append").format("iceberg").saveAsTable("logs")	支持隐藏分区，用户无需手动指定路径。
写入格式控制	.option("writeFormat", "parquet")	指定输出文件格式	.option("writeFormat", "parquet")	可选：parquet、orc、avro。
并发写入	多任务同时写入	支持高并发写入	多个 Spark 作业写入同一张表	使用乐观锁，冲突时自动重试（由 commit.retry.num 控制）。

5.2 UPSERT 实现：MERGE INTO 操作详解

子句/操作	语法	用途	代码示例	注意事项
MERGE INTO 主体	MERGE INTO target USING source ON condition	根据条件匹配源表与目标表数据	MERGE INTO local.db.users AS t USING updates AS s ON t.id = s.id	必须指定匹配条件。
WHEN MATCHED THEN UPDATE	WHEN MATCHED THEN UPDATE SET *	匹配时更新目标行	WHEN MATCHED THEN UPDATE SET *	* 表示所有字段更新为源表值。
WHEN MATCHED THEN DELETE	WHEN MATCHED THEN DELETE	匹配时删除目标行	WHEN MATCHED AND s.op = 'delete' THEN DELETE	支持条件删除。
WHEN NOT MATCHED THEN INSERT	WHEN NOT MATCHED THEN INSERT *	不匹配时插入新行	WHEN NOT MATCHED THEN INSERT *	实现 upsert 核心逻辑。
完整 MERGE 示例	综合使用	实现 UPSERT（更新或插入）	MERGE INTO local.db.users AS t USING staging_updates AS s ON t.id = s.id WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *	需启用 write.merge.mode=copy-on-write 或 merge-on-read。
性能优化建议	分区剪枝 + 小文件合并	提升 MERGE 性能	在分区字段上匹配，如 ON t.id = s.id AND t.dt = s.dt	避免全表扫描，减少 I/O。
支持的匹配条件	等值连接	当前仅支持等值条件	支持 t.a = s.a AND t.b = s.b	不支持非等值或复杂表达式。

5.3 DELETE 操作：行级删除（基于条件）

删除方式	语法	用途	代码示例	注意事项
基于条件删除	DELETE FROM table WHERE condition	删除满足条件的行	DELETE FROM local.db.users WHERE id < 100;	条件应尽量利用分区字段以提升性能。
分区级删除	DELETE FROM table WHERE partition_col IN (...)	删除整个分区数据	DELETE FROM logs WHERE dt < '2023-01-01';	推荐用于生命周期管理。
使用 DataFrame 删除	spark.sql(...)	在 Spark 中执行 SQL 删除	spark.sql("DELETE FROM local.db.users WHERE active = false")	需 Spark 3.2+ 支持。
删除限制	不支持无 WHERE 的 DELETE	必须指定删除条件	DELETE FROM table; ❌ 不支持	Iceberg 不允许全表删除语句。
底层机制	位置删除（Position Delete）或列删除（Equality Delete）	标记文件中某些行被删除	自动生成 delete files	查询时自动过滤已删除行。
性能影响	大量删除后需优化	可能产生小文件或性能下降	建议定期执行 rewrite_data_files	删除后文件未物理清除。

5.4 静默写入（Streaming Write）与连续处理支持

方法/配置	语法	用途	代码示例	注意事项
Streaming Write	df.writeStream.format("iceberg")	流式数据持续写入 Iceberg 表	df.writeStream.format("iceberg").outputMode("append").option("checkpointLocation", "/chk/loc").toTable("local.db.events")	支持 Exactly-Once 语义。
Checkpoint 位置	.option("checkpointLocation", path)	存储流式作业状态	.option("checkpointLocation", "s3a://bucket/chk/events")	必须配置，用于故障恢复。
输出模式	.outputMode("append")	仅支持 append 模式	.outputMode("append")	不支持 update/complete 模式。
微批处理（Micro-batch）	默认模式	按批次提交数据	每批生成一个新快照	低延迟，适合准实时场景。
连续处理（Continuous）	.trigger(continuous=...)	极低延迟流处理	.trigger(continuous="1 second").start()	实验性功能，需 Flink 或 Spark 3.0+。
流读取（Streaming Read）	spark.readStream.table("table")	读取 Iceberg 表的变更流	spark.readStream.table("local.db.events").filter("op = 'INSERT'")	支持变更数据捕获（CDC）场景。
小文件问题	流式写入易产生小文件	需后续合并优化	使用 rewrite_data_files 定期合并	可配置 write.batch-size 控制批次大小。

第6章：数据查询与时间旅行

6.1 SELECT 查询基础（支持 Predicate Pushdown）

查询特性	语法	用途	代码示例	注意事项
基础查询	SELECT * FROM table	查询表所有数据	SELECT id, name FROM local.db.users;	支持标准 SQL 语法。
谓词下推（Predicate Pushdown）	WHERE col = value	将过滤条件下推到文件层	SELECT * FROM logs WHERE dt = '2024-01-01' AND user_id = 100;	仅扫描匹配的文件，提升性能。
分区裁剪（Partition Pruning）	WHERE partition_col = ...	跳过不相关分区	SELECT * FROM logs WHERE dt IN ('2024-01-01', '2024-01-02');	隐藏分区同样生效。
列裁剪（Column Projection）	选择特定列	仅读取所需列	SELECT id, email FROM users;	减少 I/O，提升查询速度。
LIMIT 查询	LIMIT n	限制返回行数	SELECT * FROM users LIMIT 10;	常用于调试或抽样。
ORDER BY	ORDER BY col	排序结果	SELECT * FROM users ORDER BY id;	全局排序成本高，建议结合分区使用。
JOIN 查询	JOIN table ON ...	关联多张 Iceberg 表	SELECT a.id, b.total FROM users a JOIN orders b ON a.id = b.user_id;	支持广播、排序合并等策略。

6.2 时间旅行（Time Travel）：AS OF TIMESTAMP 与 AS OF SNAPSHOT ID

时间旅行方式	语法	用途	代码示例	注意事项
按时间戳查询	FOR SYSTEM_TIME AS OF timestamp	查询指定时间点的数据	SELECT * FROM local.db.users FOR SYSTEM_TIME AS OF '2024-01-01 00:00:00';	使用 UTC 时间。
按快照 ID 查询	FOR VERSION AS OF snapshot_id	查询指定快照的数据	SELECT * FROM local.db.users FOR VERSION AS OF 1234567890123456789;	快照 ID 可通过 table.history 获取。
查看当前快照	current_snapshot_id 函数	获取表当前快照 ID	SELECT current_snapshot_id('local.db.users');	用于记录基准版本。
时间旅行限制	快照未过期	仅能访问未被 expire_snapshots 清理的快照	EXPIRE SNAPSHOT 命令执行后无法访问	建议设置合理的保留策略。
事务一致性	基于快照隔离	查询结果为某一时刻的完整视图	不会出现部分写入的中间状态	支持一致性读。
跨分支查询	FOR VERSION AS OF branch='dev'	查询非主分支数据	SELECT * FROM table FOR VERSION AS OF branch='dev';	需 Iceberg 0.13+ 支持。

6.3 查询历史数据变更（table.history()）

字段	类型	说明	注意事项
made_current_at	TIMESTAMP	快照被设置为当前的时间	可用于审计写入时间。
snapshot_id	BIGINT	快照唯一标识符	用于 FOR VERSION AS OF 查询。
parent_id	BIGINT	父快照 ID，形成链式结构	支持回溯变更路径。
is_current_ancestor	BOOLEAN	是否为当前快照的祖先	帮助判断快照关系。

使用示例：

SELECT * FROM local.db.users.history;

| 典型用途 | 审计、调试、恢复 | 了解表的变更过程 | 结合 files 和 snapshots 表分析。 |

6.4 查看文件级统计信息（table.files()）

字段	类型	说明	注意事项
content	INT	文件内容类型（0=data, 1=position deletes, 2=equality deletes）	通常关注 data 文件。
file_path	STRING	数据文件完整路径	可用于手动检查或清理。
file_format	STRING	文件格式（PARQUET/ORC/AVRO）	由 write.format.default 决定。
partition	STRUCT	文件所属分区值	用于分析分区分布。
record_count	BIGINT	文件包含的行数	评估文件大小与查询性能。
file_size_in_bytes	BIGINT	文件字节大小	识别小文件或大文件。
column_sizes	MAP<INT, BIGINT>	每列大小（字节）	分析列存储效率。
value_counts	MAP<INT, BIGINT>	每列值数量	辅助统计分析。
null_value_counts	MAP<INT, BIGINT>	每列空值数量	数据质量检查。
lower_bounds	MAP<INT, STRING>	每列最小值	支持谓词下推优化。
upper_bounds	MAP<INT, STRING>	每列最大值	同上。

使用示例：

SELECT file_path, file_size_in_bytes, record_count
FROM local.db.users.files;

第7章：分区管理与优化

7.1 隐藏分区（Hidden Partitioning） vs 显式分区

对比维度	隐藏分区（Hidden Partitioning）	显式分区（Explicit Partitioning）	注意事项
定义方式	使用分区变换函数（如 day(ts)）	用户手动指定分区列并写入对应路径	Iceberg 推荐使用隐藏分区。
写入方式	用户写入原始数据，Iceberg 自动计算分区值	用户必须将数据写入正确分区路径（如 /dt=2024-01-01/）	显式分区易出错。
分区路径	Iceberg 自动生成（如 dt=2024-01-01）	由用户或上游系统控制	隐藏分区路径更规范。
分区演化支持	支持动态修改分区策略	不支持，需重建表	隐藏分区更灵活。
查询兼容性	支持标准 WHERE dt = '2024-01-01' 查询	同样支持	用户无感知差异。
适用场景	所有新表推荐使用	仅用于兼容旧 Hive 表或迁移场景	新项目应统一使用隐藏分区。
示例语法	PARTITIONED BY (day(ts))	PARTITIONED BY (dt STRING)	前者为隐藏，后者为显式。

7.2 分区演化（Partition Evolution）与自动迁移

操作	语法	用途	代码示例	注意事项
添加分区字段	ALTER TABLE ADD PARTITION FIELD transform	动态增加新的分区策略	ALTER TABLE logs ADD PARTITION FIELD hour(ts);	旧数据按新规则查询，无需重写。
修改分区策略	先删除再添加	更改现有分区方式	ALTER TABLE logs DROP PARTITION FIELD day(ts); ALTER TABLE logs ADD PARTITION FIELD bucket(user_id, 16);	需 Iceberg 0.13+ 支持 DROP。
分区自动迁移	无需操作	查询时自动适配多版本分区策略	SELECT * FROM logs WHERE hour(ts) = 12;	Iceberg 自动识别不同快照的分区方式。
分区演化限制	不支持直接修改字段	不能 ALTER PARTITION FIELD	必须通过删除+添加实现	保留历史快照兼容性。
查询一致性	跨分区策略查询	支持统一 SQL 查询历史数据	同一 SQL 可查询旧 day(ts) 和新 hour(ts) 数据	提升使用便利性。
元数据版本控制	每次演化生成新 metadata 文件	记录分区策略变更历史	metadata 文件保留 previous-versions-max 个	可用于审计与恢复。

7.3 使用 Partition Transform 函数（year, month, day, hour, bucket, truncate）

分区函数	语法	用途	代码示例	注意事项
year	year(ts)	按年份分区	PARTITIONED BY (year(event_time))	分区名：year=2024。
month	month(ts)	按月份分区	PARTITIONED BY (month(event_time))	支持 timestamp、date 类型。
day	day(ts)	按天分区（最常用）	PARTITIONED BY (day(event_time))	等价于 date(ts)。
hour	hour(ts)	按小时分区	PARTITIONED BY (hour(event_time))	适用于高频率写入场景。
bucket	bucket(id, N)	哈希桶分区，共 N 个桶	PARTITIONED BY (bucket(user_id, 32))	避免数据倾斜，提升 Join 性能。
truncate	truncate(name, len)	按前 len 个字符截断分区	PARTITIONED BY (truncate(email, 5))	适用于字符串列，控制基数。
identity	identity(col)	直接使用列值分区	PARTITIONED BY (status)	适用于低基数列。
复合分区	多个函数组合	构建多级分区	PARTITIONED BY (day(event_time), bucket(user_id, 16))	提升查询性能与管理灵活性。
转换嵌套字段	支持 struct 内字段	对嵌套列进行分区	PARTITIONED BY (day(metadata.ts))	需字段存在且类型匹配。

7.4 动态分区写入与性能影响

场景	说明	代码示例	注意事项
动态分区写入	写入时自动识别分区值，无需预定义路径	df.write.mode("append").format("iceberg").saveAsTable("logs")	Iceberg 自动提取 day(ts) 等值。
小文件问题	高频小批量写入易产生大量小文件	每批写入 100 条，每小时 10 批 → 240 个文件/天	影响查询性能与存储效率。
文件合并策略	使用 rewrite_data_files 优化	CALL system.rewrite_data_files(table => 'local.db.logs')	建议定期调度执行。
批次大小建议	控制每批写入行数	每批 10K~100K 行	平衡延迟与文件大小。
分区设计影响	过细分区增加文件数	hour(ts) + bucket(id, 64) → 高基数	避免”过度分区”。
写入性能优化	预分区 + 批处理	使用 repartition(32) 调整并行度	减少 Spark task 数量。
流式写入调优	设置 checkpoint 与触发间隔	.trigger(processingTime='5 minutes')	控制提交频率，减少快照数量。
查询性能影响	小文件多 → manifest 文件大 → 查询慢	SELECT 可能扫描数百个文件	合并后可减少至几十个。

第8章：事务与并发控制

8.1 Iceberg 的原子性保证与快照隔离

特性	说明	注意事项
原子性提交	每次写入要么全部成功，要么全部失败，不会出现中间状态。	依赖底层文件系统原子 rename（如 HDFS）。
快照隔离（Snapshot Isolation）	读操作基于固定快照进行，不会看到未提交的写入。	实现 MVCC，提升并发性能。
一致性读	事务内多次读取返回相同结果。	适用于长事务分析场景。
写不阻塞读	写入新快照时，旧快照仍可被查询。	支持高并发读写。
读不阻塞写	查询大表时不阻塞写入操作。	与传统数据库锁机制不同。
ACID 支持	提供原子性、一致性、隔离性、持久性保证。	但非传统数据库的强一致性模型。
多版本并发控制（MVCC）	每个快照为一个版本，读写操作在不同版本上进行。	避免锁竞争，提升吞吐。

8.2 并发写入冲突处理机制（乐观锁）

机制	说明	注意事项
乐观锁（Optimistic Locking）	不加锁，提交时检查元数据版本是否匹配。	冲突时失败，由客户端重试。
提交冲突检测	写入前获取最新 metadata 文件，提交时校验其未被修改。	基于文件系统版本或 ETag。
冲突场景	两个作业同时写入同一张表。	后提交者检测到版本不一致。
冲突处理策略	自动重试（由配置控制）。	默认重试 4 次（commit.retry.num）。
重试间隔	指数退避算法。	避免持续冲突，如 1s, 2s, 4s…
配置参数	commit.retry.num 和 commit.retry.interval。	可在 catalog 中设置。
高并发建议	减少写入频率或分区隔离。	不同作业写入不同分区可避免冲突。

8.3 事务提交流程与 Commiter 角色

阶段	说明	注意事项
1. 开始事务	读取当前表 metadata 文件，获取当前 snapshot-id。	作为本次提交的父快照。
2. 写入数据文件	将数据写入临时路径（如 _temp/）。	文件未提交前不可见。
3. 生成 Manifest 文件	记录本次写入的所有数据文件信息。	包含统计信息用于谓词下推。
4. 提交元数据变更	将新 metadata 文件写入 metadata/ 目录，并原子替换当前指针。	核心步骤，保证原子性。
5. 清理临时文件	删除临时写入的数据文件（如有）。	确保无残留。
Committer 角色	负责执行提交逻辑的组件。	在 Spark 中由 SparkCommitter 实现。
原子性保证	依赖文件系统 rename 操作。	S3 需启用 s3a.committer 实现类。
提交失败处理	临时文件保留，metadata 未更新。	可安全重试。

8.4 处理写入失败与回滚策略

故障类型	处理机制	注意事项
任务失败（Task Failure）	Spark 自动重试 task。	不影响事务，由执行引擎处理。
作业失败（Job Failure）	未提交的写入不生效。	临时文件可被清理，无数据残留。
提交冲突	乐观锁检测失败。	客户端应捕获异常并重试。
网络中断	写入中断，临时文件未提交。	不影响表状态，可重试。
存储不可写	无法写入数据或元数据。	报错终止，需人工干预。
手动回滚	使用时间旅行 + 覆盖写入。	INSERT OVERWRITE ... FOR VERSION AS OF snapshot_id。
元数据损坏	使用 metadata-log 恢复。	从历史 metadata 文件恢复。
建议实践	设置合理的重试策略。	配置 commit.retry.num=5。
监控告警	监控写入失败率、重试次数。	及时发现并发冲突或配置问题。

第9章：元数据管理与维护

9.1 元数据文件清理策略（expire_snapshots）

方法/参数	语法	用途	代码示例	注意事项
expire_snapshots	CALL system.expire_snapshots(...)	清理过期快照，释放存储空间	CALL system.expire_snapshots(table => 'local.db.users', older_than => TIMESTAMP '2024-01-01 00:00:00', retain_last => 5)	必须保留至少 1 个快照。
older_than	older_than => timestamp	删除早于该时间的快照	older_than => CURRENT_TIMESTAMP() - INTERVAL '7' DAY	建议保留 7~30 天历史。
retain_last	retain_last => n	至少保留最近 n 个快照	retain_last => 3	防止误删当前使用版本。
表标识符	table => 'catalog.db.tbl'	指定目标表	table => 'hive_catalog.sales.orders'	必须为完整路径。
自动调度	结合 Airflow/Crontab	定期执行清理任务	每日凌晨执行 expire 策略	避免影响白天查询性能。
注意事项	-	执行后无法恢复旧快照	确保无作业依赖即将删除的快照	-

9.2 文件合并与小文件优化（rewrite_data_files）

方法/参数	语法	用途	代码示例	注意事项
rewrite_data_files	CALL system.rewrite_data_files(...)	合并小文件，提升查询性能	CALL system.rewrite_data_files(table => 'local.db.logs')	自动生成更大、更少的文件。
表标识符	table => 'catalog.db.tbl'	指定要优化的表	table => 'local.web.clicks'	支持 HadoopCatalog、HiveCatalog 等。
排除条件	where => 'partition_col = value'	仅合并特定分区	where => 'dt = ''2024-01-01'''	字符串需双引号转义。
触发场景	写入频繁的小文件表	流式写入或微批处理后	每小时/每天调度一次	建议在低峰期执行。
性能影响	I/O 密集型操作	可能短暂占用集群资源	控制并发度（如 Spark 分区数）	避免影响在线查询。
合并策略	Iceberg 自动选择	基于文件大小和数量	默认目标文件大小由 write.target-file-size-bytes 控制	可配置为 512MB 或 1GB。

9.3 统计信息清理（remove_orphan_files）

方法/参数	语法	用途	代码示例	注意事项
remove_orphan_files	CALL system.remove_orphan_files(...)	删除未被 metadata 引用的”孤儿”文件	CALL system.remove_orphan_files(table => 'local.db.users', older_than => CURRENT_TIMESTAMP() - INTERVAL '2' DAY)	回收无效存储空间。
older_than	older_than => timestamp	仅删除指定时间前的孤儿文件	older_than => NOW() - INTERVAL '3' DAY	防止误删正在写入的临时文件。
dry_run	dry_run => true	预览将被删除的文件	dry_run => true	调试时使用，不实际删除。
孤儿文件来源	写入中断、手动删除元数据等	文件存在于存储但不在 manifest 中	如 S3 上残留的 .parquet 文件	占用空间且无引用。
执行频率	建议每周或每月一次	配合 expire_snapshots 使用	先 expire，再 remove_orphan	确保快照已过期后再清理文件。
注意事项	务必设置 older_than	避免误删活跃文件	-	-

9.4 元数据刷新（refresh_table）与缓存控制

方法/参数	语法	用途	代码示例	注意事项
refresh_table	CALL system.refresh_table('table')	强制刷新表的元数据缓存	CALL system.refresh_table('local.db.sales')	适用于跨会话变更感知。
缓存机制	Spark / Trino 自动缓存 metadata	提升重复查询性能	SELECT * FROM t LIMIT 10; 第二次更快	但可能读到旧 schema。
何时需要刷新	其他会话修改了表结构或分区	如 ALTER TABLE 添加列	当前会话仍显示旧 schema	必须手动 refresh。
外部写入感知	Flink、Spark Streaming 写入后	查询端需刷新才能看到最新数据	在 Spark SQL 中执行 refresh_table	尤其在共享表场景下重要。
自动刷新配置	spark.sql.catalog.<name>.cache-enabled=false	关闭缓存（不推荐）	降低性能，避免 stale 数据	一般保持开启，按需刷新。
Trino 中刷新	CALL system.refresh_table(...)	同样支持	CALL system.refresh_table('iceberg_db.logs');	多引擎环境下通用。

第10章：高级特性与扩展功能

10.1 分支（Branch）与标签（Tag）的使用场景与操作

概念	语法	用途	代码示例	注意事项
创建分支	CALL system.create_branch(...)	创建独立开发/测试环境	CALL system.create_branch(table => 'local.db.users', branch => 'dev', snapshot_id => 123...)	基于某个快照创建。
创建标签	CALL system.create_tag(...)	标记重要版本（如发布点）	CALL system.create_tag(table => 'sales.fact', tag => 'v1.0-release', snapshot_id => 456...)	不可变标记，用于审计。
查询分支数据	FOR VERSION AS OF branch='xxx'	读取分支上的最新状态	SELECT * FROM local.db.users FOR VERSION AS OF branch='dev';	类似 Git 的 branch 查看。
时间旅行到标签	FOR VERSION AS OF tag='xxx'	固定查询某发布版本数据	SELECT * FROM fact.sales FOR VERSION AS OF tag='v1.0-release';	保证分析一致性。
删除分支/标签	DROP BRANCH/TAG	清理不再使用的分支	DROP BRANCH local.db.users.dev;	不影响主分支数据。
使用场景	A/B 测试、灰度发布、数据修复隔离	避免污染主分支	在 dev 分支测试 ETL 脚本	提高数据安全性。
注意事项	需 Iceberg 0.13+ 且 Catalog 支持	REST Catalog 推荐	Hive Metastore 不支持	建议生产环境启用。

10.2 行级变更日志（Changelog / Row-level Deletes & Updates）

特性	说明	注意事项
行级删除支持	支持 DELETE FROM t WHERE ...	生成 position delete 文件。
行级更新模拟	通过 MERGE INTO 实现 upsert	支持字段级更新。
Changelog 生成	启用后记录每条记录的操作类型（I/U/D）	用于 CDC 场景。
查询变更流	read_changelog 函数	读取表的插入、更新、删除事件。
存储开销	产生额外 delete files	增加存储成本。
适用场景	实时数仓同步、审计追踪、事件溯源	替代传统数据库 binlog。

10.3 向量化读取与谓词下推优化

优化技术	说明	注意事项
向量化读取（Vectorized Reading）	一次处理一批行（如 4096），减少函数调用开销	Parquet/ORC 原生支持。
谓词下推（Predicate Pushdown）	将 WHERE 条件下推到文件扫描层	只读取满足条件的文件。
列裁剪（Column Projection）	仅读取 SELECT 中涉及的列	减少 I/O 量。
分区裁剪（Partition Pruning）	跳过不匹配的分区目录	大幅减少扫描范围。
统计信息利用	利用 lower_bounds / upper_bounds 过滤文件	如 WHERE id > 1000 跳过 id_max < 1000 的文件。
向量化执行引擎	Spark AQE、Trino 并行扫描	提升整体查询吞吐。

10.4 支持 Z-Order、Bloom Filter 等索引加速查询

索引类型	配置方式	用途	注意事项
Bloom Filter	ALTER TABLE SET TBLPROPERTIES ('write.bloom-filter.columns'='col_name')	加速高基数列的等值查询	如 id = 123。
Z-Order 排序	ALTER TABLE SET TBLPROPERTIES ('write.zorder.enabled'='true', 'write.zorder.cols'='col1,col2')	多维数据局部性优化	提升 (col1=A AND col2=B) 查询性能。
前缀索引（自动）	基于列统计	自动跳过大量文件	无需配置，Iceberg 默认行为。
数据排序（Sort Order）	ADD SORT ORDER	指定数据物理排序	提升范围查询效率。
使用建议	根据查询模式选择	避免过度索引	每增加一个索引都带来写入成本。
查询效果	显著减少文件扫描数量	提升 2x~10x 查询速度	尤其在大表上效果明显。
局限性	不支持所有数据类型	如复杂类型（array、map）不支持 bloom filter	需合理设计表结构。

第11章：与其他系统的集成

11.1 Flink + Iceberg 实时数仓集成

配置项/操作	语法	用途	代码示例	注意事项
添加依赖	flink-iceberg-connector	集成 Flink 与 Iceberg	Maven: org.apache.iceberg:iceberg-flink-runtime-1.17	版本需匹配 Flink 版本。
创建 Catalog	CREATE CATALOG ... TYPE iceberg	在 Flink SQL 中注册 Iceberg 表源	CREATE CATALOG hive_iceberg WITH ('type'='iceberg', 'catalog-type'='hive', 'uri'='thrift://hive:9083', 'warehouse'='s3a://bucket/warehouse');	支持 HiveCatalog 或 HadoopCatalog。
流式写入	Flink DataStream API	将 Kafka 数据实时写入 Iceberg	tableEnv.executeSql("INSERT INTO hive_iceberg.db.fact_sales SELECT * FROM kafka_source");	支持 Exactly-Once。
主键去重	write.overwrite-mode=merge-on-read	实现 UPSERT 语义	'write.upsert-enabled' = 'true'	需表定义主键或唯一约束。
小文件合并	异步 compact 任务	减少小文件数量	INSERT INTO compact_job SELECT * FROM source FOR SYSTEM_TIME AS OF ...;	可配置自动合并策略。
Checkpoint 集成	Flink Checkpoint	保证写入一致性	每 5 分钟提交一次快照	与 Iceberg 快照对齐。
读取变更流	table.changelog	从 Iceberg 表读取插入/更新/删除事件	SELECT * FROM db.table.changelog;	用于 CDC 回放或审计。

11.2 Trino/Presto 查询 Iceberg 表

配置项/操作	语法	用途	代码示例	注意事项
Catalog 配置	etc/catalog/iceberg.properties	注册 Iceberg 数据源	connector.name=iceberg hive.metastore.uri=thrift://hive:9083 warehouse=s3a://bucket/wh	支持 HMS 或 REST Catalog。
基础查询	SELECT 语句	查询 Iceberg 表数据	SELECT user_id, amount FROM iceberg.db.sales WHERE dt = '2024-01-01';	完全兼容标准 SQL。
时间旅行查询	FOR VERSION AS OF / AS OF TIMESTAMP	查看历史版本数据	SELECT * FROM iceberg.db.users FOR TIMESTAMP AS OF '2024-01-01 00:00:00';	支持审计与回滚分析。
分支与标签查询	FOR BRANCH 'dev'	读取非主分支数据	SELECT * FROM t FOR BRANCH 'dev';	需 Iceberg 0.13+。
性能优化	分区裁剪 + 谓词下推	提升查询效率	WHERE 条件中使用分区字段	Trino 自动下推过滤条件。
并发控制	多用户同时查询	支持高并发 OLAP 查询	数百个并发查询无锁冲突	MVCC 保证一致性。
元数据访问	table.files, history	分析表健康状态	SELECT * FROM t.files; SELECT * FROM t.history;	用于监控和治理。

11.3 Hive 兼容模式访问 Iceberg

方式	语法	用途	代码示例	注意事项
Hive Catalog 模式	catalog-type=hive	使用 Hive Metastore 管理 Iceberg 表	spark.sql.catalog.hive_catalog.type = hive	最常用兼容方式。
Hive 读取 Iceberg 表	Hive 3.0+	直接查询 Iceberg 表	SELECT * FROM iceberg_table;	需添加 Iceberg Hive Runtime JAR。
表共享	Spark/Flink 写入，Hive 查询	统一元数据层	Spark 写入 Iceberg，Hive 做 BI 报表	避免 Hive 直接修改。
DDL 兼容性	支持 SHOW TABLES, DESCRIBE	基本元数据操作	SHOW TABLES IN iceberg_db;	不支持 ALTER COLUMN 等高级操作。
写入限制	Hive 不支持复杂写入	仅限简单 INSERT	INSERT INTO t VALUES (...);	不推荐 Hive 写入。
分区同步	自动识别隐藏分区	Hive 可按 dt='2024-01-01' 过滤	SELECT * FROM logs WHERE dt='2024-01-01';	分区值由 Iceberg 自动生成。
推荐场景	BI 工具通过 Hive 访问	如 Tableau、Superset	通过 HiveServer2 暴露接口	保持写入端统一。

11.4 Kafka + Flink + Iceberg 实时入湖案例

组件	角色	配置/代码示例	注意事项
Kafka	数据源	主题：user_events，格式：JSON/Avro	保证高吞吐、持久化。
Flink	流处理引擎	StreamExecutionEnvironment env = ...; KafkaSource source = KafkaSource.builder()...;	设置 checkpoint 间隔。
Schema 注册	Confluent Schema Registry	Avro 格式自动解析 schema	与 Iceberg 表结构映射。
Flink SQL 建表	定义源表和目标表	CREATE TABLE kafka_source (user_id BIGINT, event_type STRING, ts TIMESTAMP(3)) WITH ('connector' = 'kafka', ...); CREATE TABLE iceberg_sink (...) PARTITIONED BY (day(ts)) LOCATION 's3a://lake/dw/events';	分区策略提前设计。
写入 Iceberg	插入流数据	INSERT INTO iceberg_sink SELECT * FROM kafka_source;	支持 append 和 upsert 模式。
UPSERT 模式	去重更新	'write.upsert-enabled' = 'true'	需启用 merge-on-read。
小文件治理	Compact 作业	单独启动 compact 任务合并文件	或配置自动 compact 策略。
端到端延迟	通常 1~5 分钟	受 checkpoint 间隔影响	可优化至秒级。
生产建议	分区按天+桶分组	PARTITIONED BY (day(ts), bucket(user_id, 64))	防止数据倾斜。

第12章：生产实践与性能调优

12.1 小文件问题成因与治理方案

成因	说明	治理方案	注意事项
微批流式写入	每 1 分钟提交一批 → 每天 1440 个文件	增大触发间隔：5~10 分钟	平衡延迟与文件数。
高频批处理	每小时一个作业写入	合并多个作业输出	或使用队列缓冲。
Spark 并行度高	1000 个 task → 1000 个文件	repartition(n) 控制输出文件数	n 建议为预期文件大小 / 目标大小。
动态分区写入	分区基数高（如 user_id）	使用 bucket(user_id, N) 替代	控制最大分区数。
MERGE INTO 更新	每次生成新文件	启用 write.merge.mode=copy-on-write	减少位置删除文件。
治理工具	rewrite_data_files	CALL system.rewrite_data_files(...)	定期调度执行。
预防策略	写入前预分区	df.repartition(32).write...	输出文件大小控制在 128MB~1GB。

12.2 分区设计最佳实践

原则	建议	示例	注意事项
分区粒度适中	避免过细或过粗	按 day(ts) 而非 hour(ts) 或 year(ts)	每个分区文件建议 512MB~1GB。
高频查询字段优先	将 WHERE 常用字段作为分区	如 status, region	但避免高基数列。
使用隐藏分区	推荐 day(ts), bucket(id, 32)	PARTITIONED BY (day(event_time), bucket(user_id, 32))	自动管理路径。
避免字符串直接分区	易导致数据倾斜	不要用 email 直接分区	可用 truncate(email, 3) 或 bucket。
时间分区必选	几乎所有表都应包含时间维度	day(ts) 作为一级分区	便于生命周期管理。
分区演化支持	设计初期不必完美	后续可用 ADD PARTITION FIELD 调整	旧数据自动适配。
复合分区顺序	高选择性字段在前	day(ts) 在前，bucket(id) 在后	提升谓词下推效率。

12.3 查询性能优化建议（分区裁剪、列裁剪、索引）

优化手段	说明	使用方式	注意事项
分区裁剪	跳过不相关分区目录	WHERE dt = '2024-01-01'	对隐藏分区同样生效。
列裁剪	仅读取 SELECT 涉及的列	SELECT id, name FROM t;	减少 I/O，提升速度。
谓词下推	将过滤条件下推到文件层	WHERE price > 100 AND cat = 'A'	依赖 manifest 统计信息。
Bloom Filter	加速等值查询	ALTER TABLE SET TBLPROPERTIES ('write.bloom-filter.columns'='id')	仅对 = 有效，增加写入开销。
Z-Order	提升多维查询效率	write.zorder.cols='x,y,z'	适用于地理、标签等场景。
文件大小优化	大文件减少 manifest 扫描	目标 512MB~1GB	太大会降低并行度。
向量化读取	批量处理数据	Parquet/ORC 原生支持	Spark/Trino 默认开启。
AQE 优化	Spark 动态调整执行计划	spark.sql.adaptive.enabled=true	推荐生产环境开启。

12.4 监控与审计：通过元数据表分析表健康状态

元数据表	查询语句	用途	注意事项
table.history	SELECT * FROM t.history;	查看快照变更历史	识别频繁写入或异常作业。
table.files	SELECT count(*) FROM t.files;	统计文件数量	文件数 > 1000 视为小文件严重。
table.files（大小）	SELECT sum(file_size_in_bytes)/1024/1024 AS size_mb FROM t.files;	计算总存储占用	结合分区分析空间分布。
table.files（记录数）	SELECT avg(record_count) FROM t.files;	评估平均文件行数	过低说明存在大量小文件。
table.manifests	SELECT * FROM t.manifests;	查看 manifest 文件详情	分析分区统计信息有效性。
table.partitions	SELECT * FROM t.partitions;	查看当前分区统计	包括行数、空值、边界值等。
监控指标	文件数、大小、快照频率	设置告警阈值	如每日新增文件 > 5000 触发告警。
审计用途	结合 history 与 files	追踪数据变更来源	用于合规性检查。