4. Flink SQL 创建 Iceberg 表：语法、分区与主键
这一节我们直接上手干活。用 Flink SQL 在 Iceberg 上建表，说白了就是把 Hive 那套 DDL 搬到 Flink 里，但加了不少 Iceberg 自己的东西。我个人习惯先把表建出来跑通，再回头调优，这样心里有底。
4.1 基础 CREATE TABLE 语法
先看一个最简版本。你打开 Flink SQL Client，敲下面这段：
CREATE TABLE iceberg_catalog.my_db.log_events (
event_id STRING,
event_time TIMESTAMP(3),
user_id STRING,
event_type STRING,
event_data STRING
) WITH (
'connector' = 'iceberg',
'catalog-type' = 'hadoop',
'warehouse' = 'hdfs://namenode:8020/warehouse'
);
嗯，这里要注意：catalog-type 我推荐用 hadoop 或 hive。如果你公司已经有 Hive Metastore，直接用 hive 类型，省事。我在项目中遇到过有人用 hadoop 类型但没配 HDFS 权限，结果建表成功但写数据报错——后来发现是 warehouse 路径没写对。
SHOW CATALOGS 确认你的 Iceberg catalog 已经注册到 Flink 里了。我经常看到新人直接建表，然后报错说 catalog 不存在。
4.2 分区策略：怎么分才不后悔
分区是 Iceberg 的核心能力之一。它跟 Hive 分区最大的区别是：Iceberg 隐藏了分区目录，你不需要在 SQL 里写 WHERE partition_col = xxx，Iceberg 会自动做分区裁剪。
语法上，分区定义写在 PARTITIONED BY 里：
CREATE TABLE iceberg_catalog.my_db.user_actions (
action_id STRING,
user_id STRING,
action_time TIMESTAMP(3),
action_type STRING,
action_data STRING
) PARTITIONED BY (
days(action_time),
action_type
) WITH (
'connector' = 'iceberg',
'catalog-type' = 'hadoop',
'warehouse' = 'hdfs://namenode:8020/warehouse'
);
这里 days(action_time) 是 Iceberg 的隐藏分区转换——它不会真的在表里加一列，而是按 action_time 的天做分区。你想想看，如果按原始时间戳分区，每个毫秒一个分区，那还得了？
常用的分区转换有这些：
| 转换函数 | 适用类型 | 说明 |
|---|---|---|
years(ts) |
时间戳/日期 | 按年分区，适合历史归档 |
months(ts) |
时间戳/日期 | 按月分区，我常用在日志表 |
days(ts) |
时间戳/日期 | 按天分区，最常用 |
hours(ts) |
时间戳 | 按小时分区，数据量极大时用 |
bucket(n, col) |
任意类型 | 哈希分桶，解决数据倾斜 |
truncate(len, col) |
字符串 | 截断前 N 个字符做分区 |
user_id 直接做分区，结果用户量一大，分区数爆炸，元数据查询慢得离谱。后来改成 bucket(100, user_id)，问题解决。记住：分区数控制在几百到几千最合适，别搞出几十万个分区。
4.3 主键定义：UPSERT 的关键
Iceberg 从 v2 开始支持主键。主键的作用不是强制唯一约束，而是告诉 Iceberg：当数据进来时，按这些字段去重或更新。
语法很简单，在列定义后面加 PRIMARY KEY：
CREATE TABLE iceberg_catalog.my_db.user_profile (
user_id STRING,
user_name STRING,
email STRING,
update_time TIMESTAMP(3),
status STRING,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'iceberg',
'catalog-type' = 'hadoop',
'warehouse' = 'hdfs://namenode:8020/warehouse',
'write.upsert.enabled' = 'true'
);
注意 NOT ENFORCED 这个关键字。它表示 Iceberg 不会在写入时检查主键唯一性——检查成本太高了。你想想看，如果每次写入都要全表扫描去重，那还怎么玩？所以 Iceberg 选择信任上游，由 Flink 或业务层保证数据不重复。
另外，write.upsert.enabled 必须设为 true，否则主键不会生效。我见过有人配了主键但忘了开这个参数，结果数据一直追加，没有去重——排查了半天才发现。
4.4 完整示例：一张实战表
把上面三个知识点串起来，给你一个我在生产环境用过的表结构：
CREATE TABLE iceberg_catalog.ods.order_events (
order_id STRING,
user_id STRING,
product_id STRING,
amount DECIMAL(10, 2),
order_status STRING,
event_time TIMESTAMP(3),
proc_time AS PROCTIME(),
PRIMARY KEY (order_id) NOT ENFORCED
) PARTITIONED BY (
days(event_time)
) WITH (
'connector' = 'iceberg',
'catalog-type' = 'hive',
'warehouse' = 'hdfs://namenode:8020/warehouse/ods',
'write.upsert.enabled' = 'true',
'write.format.default' = 'parquet',
'write.target-file-size-bytes' = '134217728'
);
这里我加了几个生产参数：
write.format.default设成 parquet，压缩比高，查询快。write.target-file-size-bytes设成 128MB，避免小文件问题。这个值我调过很多次，128MB 是性价比比较高的。proc_time是处理时间，用于 Flink 的窗口计算，跟 Iceberg 本身无关，但经常一起用。
4.5 知识体系总览
下面这张图帮你理清 Flink SQL 创建 Iceberg 表的核心逻辑：
从这张图你可以看到，建表不是简单敲一行 DDL 就完事。列定义、分区策略、主键、WITH 参数，每个环节都有讲究。我个人建议你从简单开始，先不加分区和主键，跑通基础链路，再逐步加上这些特性。