本文主要面向需要使用 Paimon 进行实时数据湖开发的工程师，为工程师提供表设计和参数调优的建议，包含：

• 从 0 到 1 的建 Catalog / 建库 / 建表 SQL
• 常见表模型（Append / 主键 Upsert / Partial Update / Aggregation）的模板与选型建议
• 分区、分桶（bucket/bucket-key）设计建议
• 关键参数：merge-engine、changelog-producer、metastore.partitioned-table、TTL、快照保留、compaction 相关
• 常见问题与排障提示

你可以直接复制本手册中的 SQL 到 Flink SQL 控制台，再替换占位符为真实值。

1. 快速决策（先选表模型） #

优先回答三个问题：

1. 是否需要按主键更新（Upsert）？
2. 是否需要“只更新部分列”（宽表逐步补齐）？
3. 是否需要“实时聚合统计”（sum/cnt 等）？

对应推荐如下：

2. Catalog 与访问路径 #

Paimon 推荐使用三段式访问路径：${catalog}.${db}.${table}。你需要先创建 Catalog，然后创建 Database，再创建 Table。

2.1 FileSystem Catalog（不依赖 Metastore） #

CREATE CATALOG paimon_catalog WITH (
  'type' = 'paimon',
  'warehouse' = 'tos://<tos-bucket>/paimon/warehouse'
);

USE CATALOG paimon_catalog;

2.2 LAS Catalog（Hive metastore + is-las） #

CREATE CATALOG paimon_las_catalog WITH (
  'type' = 'paimon',
  'metastore' = 'hive',
  'is-las' = 'true',
  'hive.client.las.region.name' = '<region>',
  'hive.metastore.uris' = 'thrift://lakeformation.las.<region>.ivolces.com:48869',
  'hive.hms.client.is.public.cloud' = 'true',
  'hive.client.las.ak' = '<ak>',
  'hive.client.las.sk' = '<sk>',
  'catalog.properties.metastore.catalog.default' = '<las-catalog>',
  'warehouse' = 'tos://<tos-bucket>/paimon/warehouse'
);

USE CATALOG paimon_las_catalog;

说明：

• 建议 AK、SK 等变量采用平台的变量管理功能，避免敏感信息明文写在代码中造成泄露
• 使用 LAS Catalog 时，Validate/解析阶段可能触发 metastore 校验，偶发失败不一定是 SQL 语法问题（见 FAQ）。

2.3 数据目录功能 #

使用 Flink 平台提供的数据目录功能，则不需要在 SQL 中使用创建 Catalog 的语法。

直接使用三段式对 Paimon 数据表进行访问，或者直接 USE Catalog 即可。

USE CATALOG paimon_las_catalog;

3. Database（建议显式创建） #

CREATE DATABASE IF NOT EXISTS test_db;
USE test_db;

说明：

• 某些环境下启动时报 DatabaseNotExistException，通常是因为未显式创建 DB；建议始终在 SQL 中写 CREATE DATABASE IF NOT EXISTS ...。

4. 表模型与模板 #

4.1 Append 表（只追加，不更新） #

适用：

• 日志明细、事件流落地
• 不需要根据主键回写

推荐：

• changelog-producer = none
• 不需要设置主键
• 分区：常用 dt（按天）；吞吐更大可 dt + hh

模板：

CREATE TABLE IF NOT EXISTS append_table (
  id BIGINT,
  name STRING,
  amount DECIMAL(10, 2),
  event_time TIMESTAMP(3),
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'changelog-producer' = 'none'
);

4.2 主键表（非分区） #

适用：

• 维表、状态表、结果表（按主键 Upsert）

推荐：

• 定义主键：PRIMARY KEY (...) NOT ENFORCED
• 通常设置 bucket
• changelog-producer：
  • 下游需要流读完整变更：lookup (建议) 或 full-compaction
  • 仅需要输入语义，例如上游如果已经是 CDC 数据：input

模板：

CREATE TABLE IF NOT EXISTS primary_key_table (
  word STRING,
  cnt BIGINT,
  PRIMARY KEY (word) NOT ENFORCED
) WITH (
  'bucket' = '4',
  'changelog-producer' = 'input'
);

4.3 主键表（分区） #

适用：

• 既要按分区管理生命周期，又要按主键更新

硬性建议：

• 主键字段必须包含所有分区字段（否则会出现语义/一致性问题）
• 若你使用 LAS/HMS 并希望分区同步：设置 metastore.partitioned-table = true

模板：

CREATE TABLE IF NOT EXISTS primary_key_partitioned (
  word STRING,
  cnt BIGINT,
  dt STRING,
  hh STRING,
  PRIMARY KEY (dt, hh, word) NOT ENFORCED
) PARTITIONED BY (dt, hh)
WITH (
  'bucket' = '4',
  'changelog-producer' = 'input',
  'metastore.partitioned-table' = 'true'
);

4.4 Partial Update（宽表模式：只更新部分列） #

适用：

• 部分 CDC 数据源仅包含更新列，不包含全部列
• 宽表逐步补齐字段
• 嵌套字段/数组需要按 key 合并更新

推荐组合：

• merge-engine = partial-update
• changelog-producer = lookup（或 full-compaction，非必要请优先选择 lookup）
• 必须定义主键
• 通过字段级参数定义合并规则（sequence-group / nested_update / nested-key 等）

模板：

CREATE TABLE IF NOT EXISTS partial_update_table (
  uid INT,
  username STRING,
  reg_time TIMESTAMP(3),
  logintypes ARRAY<ROW<logintype STRING, bind_time TIMESTAMP(3)>>,
  last_bind_time TIMESTAMP(3),
  vip_is_valid BOOLEAN,
  vip_start_time TIMESTAMP(3),
  vip_end_time TIMESTAMP(3),
  PRIMARY KEY (uid) NOT ENFORCED
) WITH (
  'bucket' = '4',
  'merge-engine' = 'partial-update',
  'changelog-producer' = 'lookup',
  'fields.last_bind_time.sequence-group' = 'logintypes',
  'fields.logintypes.aggregate-function' = 'nested_update',
  'fields.logintypes.nested-key' = 'logintype'
);

4.5 Aggregation（实时聚合） #

适用：

• 窗口聚合结果落地
• 指标汇总表

推荐组合：

• merge-engine = aggregation
• changelog-producer = lookup（或 full-compaction）
• 对参与聚合的字段配置 aggregate-function
• 对 retract 行为敏感时，按需设置 ignore-retract

模板：

CREATE TABLE IF NOT EXISTS aggregate_table (
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  category STRING,
  item_id BIGINT,
  total_amount DECIMAL(10, 2),
  cnt BIGINT,
  dt STRING,
  PRIMARY KEY (window_start, window_end, category, item_id, dt) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
  'bucket' = '4',
  'changelog-producer' = 'lookup',
  'merge-engine' = 'aggregation',
  'fields.total_amount.aggregate-function' = 'sum',
  'fields.cnt.aggregate-function' = 'sum',
  'fields.total_amount.ignore-retract' = 'true',
  'fields.cnt.ignore-retract' = 'true'
);

5. 分区设计建议（dt / hh） #

常用方案：

• 按天：dt STRING，常用格式 yyyy-MM-dd
• 按小时：dt STRING, hh STRING

建议：

• 分区主要用于“裁剪扫描范围 + 生命周期管理”，不是提升单分区写入吞吐
• 分区字段设计变更（例如从 dt -> dt+hh）通常需要清理旧数据与旧元数据后重建
• 分区主键表必须把分区字段放进主键里

6. 分桶设计建议（bucket / bucket-key） #

什么时候需要分桶：

• 主键表、partial-update、aggregation 表：通常建议设置 bucket
• 数据量大、读写并行度需求高时建议设置 bucket

经验建议：

• 单个 bucket 的数据量尽量控制在约 1-2GB
• bucket：分桶数（例如 4 / 8 / 16），注意这个参数在两种表类型间有一些语义的区别。
  • Append 表：默认值为 -1 。代表不分桶，行为接近传统 Hive 表结构。
  • 主键表：默认值为 -1 。代表动态分桶表，动态分桶表不支持并发写入，也有额外性能损耗，生产不推荐使用。
• bucket-key：分桶键
  • Append 表：Paimon 1.1+ 版本，如果设置了 bucket，则必须设置 bucket-key。
  • 主键表：默认是主键 HASH 进行分桶。可以按照业务需求进行设置分桶字段。

7. Changelog Producer 说明 #

提示：

• partial-update、aggregation 的流读场景通常要搭配 lookup （较为常见）或 full-compaction（不推荐使用）。

8. 生命周期与常用参数（TTL/快照/压缩） #

8.1 忽略删除事件 #

'ignore-delete' = 'true'

8.2 非分区表：行级过期 #

对于非分区表，建议使用行级过期时间进行数据清理。

'record-level.expire-time' = '30 d',
'record-level.time-field' = 'update_time'

8.3 分区过期（按分区值时间） #

为了降低存储成本，建议设置分区过期时间。

'partition.expiration-strategy' = 'values-time',
'partition.expiration-time' = '7 d',
'partition.expiration-check-interval' = '1 d',
'partition.timestamp-formatter' = 'yyyy-MM-dd',
'partition.timestamp-pattern' = '$dt'

8.4 快照保留 #

为了避免数据延迟过高导致消费不及时，建议调高快照保留时间。

'snapshot.time-retained' = '3 d',
'snapshot.expire.execution-mode' = 'async'

8.5 异步文件合并（Compaction 相关） #

如果你不希望文件合并阻塞 Flink 写入 Paimon 表，那么可以使用异步文件合并的方式解耦相关过程。

'num-sorted-run.stop-trigger' = '2147483647',
'sort-spill-threshold' = '5',
'changelog-producer.lookup-wait' = 'false',

说明：compaction 参数在不同 Paimon/Flink 版本上差异较大，且很多属于“高级调优项”。建议优先使用平台默认值；确需调优时，以你当前平台的 Paimon 参数列表与官方文档为准再配置。具体 Compaction 原理和参数参考：[客户] Paimon Compaction 原理和最佳实践

8.6 上下游主键不一致（sink upsert materialize） #

当上游写入语义与下游主键约束不一致时，如果下游 Paimon 表已经设置了合理的排序键，可考虑在任务参数中关闭 upsert materialize：

table.exec.sink.upsert-materialize: NONE

9. 最小可运行示例：Kafka -> Paimon（Append + dt 分区） #

9.1 Kafka Source（JSON） #

CREATE TABLE kafka_source (
  id BIGINT,
  name STRING,
  amount DECIMAL(10, 2),
  event_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = '<topic>',
  'properties.bootstrap.servers' = '<brokers>',
  'properties.group.id' = '<group-id>',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors' = 'true'
);

9.2 Paimon Sink（Append 表） #

CREATE TABLE IF NOT EXISTS paimon_append_sink (
  id BIGINT,
  name STRING,
  amount DECIMAL(10, 2),
  event_time TIMESTAMP(3),
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'changelog-producer' = 'none'
);

9.3 写入（派生 dt 分区） #

INSERT INTO paimon_append_sink
SELECT
  id,
  name,
  amount,
  event_time,
  DATE_FORMAT(event_time, 'yyyy-MM-dd') AS dt
FROM kafka_source;

10. 常见误区（避免踩坑） #

10.1 把warehouse 当成表参数 #

warehouse 是 Catalog 级别 的属性（在 CREATE CATALOG ... WITH (...) 里配置），不是表级参数。建完 Catalog 后，通过 USE CATALOG ...; 再去建库建表即可。

10.2 在 Paimon 表 DDL 里写connector/path/... #

在当前“Catalog + Database + Table”的推荐工作流里，Paimon 表本身通常不需要写 connector='paimon'、path 之类“旧式直连口径”。正确方式是：

1. CREATE CATALOG ... 指定 metastore/warehouse
2. USE CATALOG ...
3. CREATE DATABASE ...
4. CREATE TABLE ... WITH (...) 里只放 Paimon 表属性（如 bucket、merge-engine、changelog-producer、字段级合并规则等）

10.3 Append 表默认不要写bucket #

Append 表通常不需要 bucket（除非你明确要提升读写并行度，或者使用 append queue 的能力）。如果需要分桶，请结合数据量与查询模式设置合理 bucket 数，并需要设置 bucket-key （1.1+ 版本强制需求）。

11. FAQ / 常见问题 #

11.1 Validate 阶段出现org.apache.thrift.transport.TTransportException #

原因：

• Validate 不仅做语法解析，还可能触发对 connector/catalog/metastore 的连通性与可用性校验
• 使用 LAS Catalog 时，可能访问 LAS/HMS 失败（网络/权限/服务状态）

建议：

• 先确认网络与权限
• 将 Validate 视为“最佳努力”的预检查：失败不等价于 SQL 语法错误

11.2 启动时报DatabaseNotExistException #

原因：

• 静态解析 SQL 时不会自动去 metastore 拉取已有 database

处理：

CREATE DATABASE IF NOT EXISTS your_db;
USE your_db;