数据湖构建：Paimon主键表：配置与特性

数据的分发

默认情况下，DLF将根据每条数据主键的哈希值，确定该数据属于哪个分桶。

数据可见性

延迟分桶表的数据可见性取决于写入方式。满足以下条件时，DLF 将启用直写固定分桶优化（即跳过 bucket-postpone 临时目录，将数据直接写入固定分桶），数据写入并 commit 后即刻可见：

• 主键表未启用 deletion vectors，且通过 Flink 批作业或 Spark 作业执行 INSERT INTO 或 INSERT OVERWRITE 操作。

• 主键表已启用 deletion vectors，且通过 Flink 批作业或 Spark 作业执行 INSERT OVERWRITE 操作。

其它情况下，数据需等待 DLF 处理后可见，通常在 1 ~ 3 分钟内。若 DLF 正在进行资源或分桶数调整，可能延长至 5 ~ 10 分钟。可通过选择特定的存储优化模式减少调整频率，详见存储优化模式。

数据一致性

• 同一作业写入：对于同一用户作业写入的数据，DLF 优化作业将按写入顺序处理数据，即保证 sequential 级别的一致性。

  说明

  从 checkpoint 或 savepoint 重启，且并发数不变的作业，也认为与重启前是同一用户作业。

• 不同作业写入：对于不同用户作业写入的数据，Paimon 表的最终状态可能是两个作业结果数据的混合，但不会有数据丢失，即保证 snapshot isolation 级别的一致性。如果对数据的合并顺序有需求，可以设置 sequence field 以指定数据的合并顺序，详见处理乱序数据。

数据的分发

与延迟分桶一致，详见数据的分发。

数据可见性

固定分桶表的数据可见性与 deletion vectors 有关：

• 未启用 deletion vectors 的表，数据写入并 commit 之后马上可见。

• 启用 deletion vectors 的表，写入的数据需要等待 DLF 优化作业处理后才能读取。一般情况下，数据将在 1 ~ 3 分钟后可见。

数据的分发

动态分桶表会先将数据写入已有的分桶中，当分桶的数据量超过限制时，再自动创建新的分桶。以下表参数将会影响动态分桶的行为：

动态分桶表的更新行为和资源消耗取决于主键与分区键的关系：

• 主键完全包含分区键 → 非跨分区更新（仅消耗堆内存）

• 主键不完全包含分区键 → 跨分区更新（需要 RocksDB，性能开销更大）

数据可见性

与固定分桶表一致，详见数据可见性。

支持的聚合函数

• sum：求和函数。支持的数据类型为 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE。

• product：求乘积函数。支持的数据类型为 DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE。

• max：求最大值的函数。支持的数据类型为 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。

• min：求最小值的函数。支持的数据类型为 CHAR、VARCHAR、DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、TIMESTAMP_LTZ。

• first_value：返回第一次输入的值，包括 null。支持所有数据类型。

• first_not_null_value：返回第一次输入的非 null 值。支持所有数据类型。

• last_value：返回最新输入的值，包括 null。支持所有数据类型。

• last_non_null_value（默认）：返回最新输入的非 null 值。支持所有数据类型。

• listagg：将输入的字符串依次用英文逗号连接。例如，输入的字符串为 apple、banana、cherry，将返回 apple,banana,cherry。支持 STRING 类型。

• bool_and：对所有输入的值求 and。支持 BOOLEAN 类型。

• bool_or：对所有输入的值求 or。支持 BOOLEAN 类型。

• rbm32：合并多个 32 位 RoaringBitmap。支持 VARBINARY 类型。

• rbm64：合并多个 64 位 RoaringBitmap。支持 VARBINARY 类型。

• nested_update：将多行聚合到一个 ARRAY 中。通过设置表参数 'fields.<field-name>.nested-key' = 'pk0,pk1,...'，还可以将 ARRAY 中的数据根据 nested key 进行去重（保留最后一条）。若未设置 nested key，则每条新数据将追加在 ARRAY 末尾。

• nested_partial_update：与 nested_update 类似，但会根据 nested key 对 ARRAY 中的数据进行部分列更新。必须设置表参数 'fields.<field-name>.nested-key' = 'pk0,pk1,...'。

• collect：将多个 ARRAY 合并成一个。通过设置表参数 'fields.<field-name>.distinct' = 'true'，可以对 ARRAY 中的数据进行去重。

• merge_map：将多个 MAP 合并成一个。相同 key 的数据将保留最后一个 value。

聚合函数使用举例

例如，使用nested_update聚合函数合并多个子订单记录：

-- 主订单表
CREATE TABLE orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING
);

-- 子订单表
CREATE TABLE sub_orders (
  order_id BIGINT,
  sub_order_id INT,
  product_name STRING,
  price BIGINT,
  PRIMARY KEY (order_id, sub_order_id) NOT ENFORCED
);

-- 宽表
-- 将同一主订单的所有子订单聚合到 sub_orders 中
-- 同时，sub_orders 中的结果将根据 sub_order_id 进行去重
CREATE TABLE order_wide (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING,
  sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
) WITH (
  'merge-engine' = 'aggregation',
  'fields.sub_orders.aggregate-function' = 'nested_update',
  'fields.sub_orders.nested-key' = 'sub_order_id'
);

INSERT INTO order_wide

SELECT
  order_id,
  user_name,
  address,
  CAST (NULL AS ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>)
FROM orders

UNION ALL

SELECT
  order_id,
  CAST (NULL AS STRING),
  CAST (NULL AS STRING),
  ARRAY[ROW(sub_order_id, product_name, price)]
FROM sub_orders;

假设主订单表中的数据为：

• (1, 'Alice', 'add1'),

• (2, 'Bob', 'add2')

子订单表中的数据为：

• (1, 11, 'apple', 10),

• (1, 12, 'banana', 20),

• (2, 21, 'cherry', 30),

• (2, 22, 'peach', 40),

• (2, 21, 'cherry', 50)

那么宽表的最终数据为：

• (1, 'Alice', 'add1', [(11, 'apple', 10), (12, 'banana', 20)])

• (2, 'Bob', 'add2', [(21, 'cherry', 50), (22, 'peach', 40)])

为不同列分别指定合并顺序

在 partial-update 合并机制中，除了 sequence field 之外，您也可以通过 sequence group 为不同列分别指定合并顺序。该功能可用于在打宽场景下处理乱序数据，为来自不同源表的列分别指定合并顺序。

例如，考虑以下创建 Paimon 表的 Flink SQL 语句。

CREATE TABLE T (
  k INT,
  a STRING,
  b STRING,
  g_1 INT,
  c STRING,
  d STRING,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.g_2.sequence-group' = 'c,d'
);

a、b 两列以 g_1 列的值作为合并顺序依据（值越大表示数据越新），c、d 两列以 g_2 列的值作为合并顺序依据。

同时进行数据的打宽与聚合

在 partial-update 合并机制中，可以在表参数中设置 fields.<field-name>.aggregate-function，为 <field-name> 列指定聚合函数，对该列的值进行聚合。

例如，考虑以下创建 Paimon 表的 Flink SQL 语句。

CREATE TABLE T (
  k INT,
  a STRING,
  b INT,
  g_1 INT,
  c STRING,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.b.aggregate-function' = 'max',
  'fields.g_2.sequence-group' = 'c,d',
  'fields.d.aggregate-function' = 'sum'
);

a、b 两列以 g_1 列的值作为合并顺序依据（值越大表示数据越新），其中 a 列保留最新的非 null 值（未指定，默认使用last_non_null_value聚合函数），b 列保留输入的最大值。c、d 两列以 g_2 列的值作为合并顺序依据，其中 c 列保留最新的非 null 值，d 列求出输入数据的和。