实时计算Flink版：流式数据湖仓Paimon

类别

详情

支持类型

源表、维表和结果表，数据摄入目标端

运行模式

流模式和批模式

数据格式

暂不支持

特有监控指标

暂无

API种类

SQL，数据摄入YAML作业

是否支持更新或删除结果表数据

是

参数

说明

数据类型

是否必填

默认值

备注

connector

表类型。

String

否

无

• 如果在Paimon Catalog中创建Paimon表，则无需填写。

• 如果在其他Catalog中创建Paimon临时表，则固定值为paimon。

path

表存储路径。

String

否

无

• 如果在Paimon Catalog中创建Paimon表，则无需填写。

• 如果在其他Catalog中创建Paimon临时表，则为表在HDFS或OSS中的存储目录。

auto-create

创建Paimon临时表时，若指定路径不存在Paimon表文件，是否自动创建文件。

Boolean

否

false

参数取值如下：

• false（默认）：如果指定路径不存在Paimon表文件，则报错。

• true：如果指定路径不存在，则Flink系统自动创建Paimon表文件。

bucket

每个分区的分桶数。

Integer

否

1

写入Paimon表的数据将按bucket-key打散至每个bucket中。

bucket-key

分桶关键列。

String

否

无

指定将写入Paimon表的数据按哪些列的值打散至不同的Bucket中。

列名之间用英文逗号（,）分隔，例如'bucket-key' = 'order_id,cust_id'会将数据按order_id列和cust_id列的值进行打散。

changelog-producer

增量数据产生机制。

String

否

none

Paimon可以为任意输入数据流产生完整的增量数据（所有的update_after数据都有对应的update_before数据），方便下游消费者。增量数据产生机制的可选值如下：

• none（默认值）：不额外产生增量数据。下游仍然可以流读Paimon表，但读到的增量数据是不完整的（只有update_after数据，没有对应的update_before数据）。

• input：将输入数据流双写至增量数据文件中，作为增量数据。

• full-compaction：每次Full Compaction产生完整的增量数据。

• lookup：每次commit snapshot前产生完整的增量数据。

关于增量数据产生机制的选择，详情请参见增量数据产生机制。

full-compaction.delta-commits

Full Compaction最大间隔。

Integer

否

无

该参数指定了每commit snapshot多少次之后，一定会进行一次Full Compaction。

lookup.cache-max-memory-size

Paimon维表的内存缓存大小。

String

否

256 MB

该参数值会同时影响维表缓存大小和lookup changelog-producer的缓存大小，两个机制的缓存大小都由该参数配置。

merge-engine

相同primary key数据的合并机制。

String

否

deduplicate

参数取值如下：

• deduplicate：仅保留最新一条。

• partial-update：用最新数据中非null的列更新相同primary key的现有数据，其它列保持不变。

• aggregation：通过指定聚合函数进行预聚合。

关于数据合并机制的具体分析，详情请参见数据合并机制。

partial-update.ignore-delete

是否忽略delete（-D）类型的消息。

Boolean

否

false

参数取值如下：

• true：忽略delete消息。

• false：不忽略delete消息。您需要通过sequence.field等配置项来设定Sink对于delete数据的处理策略，否则可能会抛出IllegalStateException或IllegalArgumentException报错。

ignore-delete

是否忽略delete（-D）类型的消息。

Boolean

否

false

参数取值同partial-update.ignore-delete。

partition.default-name

分区默认名称。

String

否

__DEFAULT_PARTITION__

如果分区列的值为null或空字符串，将会采用该默认名称作为分区名。

partition.expiration-check-interval

多久检查一次分区过期。

String

否

1h

详情请参见如何设置分区自动过期？

partition.expiration-time

分区的过期时长。

String

否

无

当一个分区的存活时长超过该值时，该分区将会过期，默认永不过期。

一个分区的存活时长由该分区的分区值计算而来，详情请参见如何设置分区自动过期？

partition.timestamp-formatter

将时间字符串转换为时间戳的格式串。

String

否

无

设置从分区值提取分区存活时长的格式，详情请参见如何设置分区自动过期？

partition.timestamp-pattern

将分区值转换为时间字符串的格式串。

String

否

无

设置从分区值提取分区存活时长的格式，详情请参见如何设置分区自动过期？

scan.bounded.watermark

当Paimon源表产生的数据的watermark超过该值时，Paimon源表将会结束产生数据。

Long

否

无

无。

scan.mode

指定Paimon源表的消费位点。

String

否

default

详情请参见如何设置Paimon源表的消费位点？

scan.snapshot-id

指定Paimon源表从哪个snapshot开始消费。

Integer

否

无

详情请参见如何设置Paimon源表的消费位点？

scan.timestamp-millis

指定Paimon源表从哪个时间点开始消费。

Integer

否

无

详情请参见如何设置Paimon源表的消费位点？

snapshot.num-retained.max

至多保留几个最新Snapshot不过期。

Integer

否

2147483647

只要满足该配置或snapshot.time-retained其中之一，并同时满足snapshot.num-retained.min，就会触发Snapshot过期。

snapshot.num-retained.min

至少保留几个最新Snapshot不过期。

Integer

否

10

无。

snapshot.time-retained

Snapshot产生多久以后会过期。

String

否

1h

只要满足该配置或snapshot.num-retained.max其中之一，并同时满足snapshot.num-retained.min，就会触发snapshot过期。

write-mode

Paimon表的写入模式。

String

否

change-log

参数取值如下：

• change-log：Paimon表支持根据primary key进行数据的插入、删除和更新。

• append-only：Paimon表只接受数据的插入，且不支持primary key。该模式比change-log模式更加高效。

关于写入模式的具体介绍，详情请参见写入模式。

scan.infer-parallelism

是否自动推断Paimon源表的并发度。

Boolean

否

false

参数取值如下：

• true：将会根据分桶数自动推断Paimon源表的并发度。

• false：按VVP配置的默认并发。如果是专家模式就按用户配置的并发。

scan.parallelism

Paimon源表的并发度。

Integer

否

无

sink.parallelism

Paimon结果表的并发度。

Integer

否

无

sink.clustering.by-columns

指定写入Paimon结果表的聚类列。

String

否

无

对于Paimon Append Only表（非主键表），在批作业中配置该参数可以启用聚类写入功能，使数据在特定列上按大小范围聚集分布，从而提升该表的查询速度。

多个列名请使用英文逗号（,）进行分隔，例如'col1,col2'。

聚类详情请参见Apache Paimon官方文档。

sink.delete-strategy​

设定校验策略，确保系统能正确处理回撤（-D/-U）类型消息。

​​

Enum

否

NONE

校验策略取值及Sink算子应当正确处理回撤消息的行为如下：​

• ​NONE（默认值）：不做校验。​

• IGNORE_DELETE：Sink算子应当忽略-U和-D类型的消息，不发生回撤。

• NON_PK_FIELD_TO_NULL：Sink算子应当忽略-U类型的消息，但是在收到-D类型的消息时，保持主键值不变、回撤Schema中其他非主键值。

  主要用在多个Sink同时写入同一张表时部分更新的场景。​

• DELETE_ROW_ON_PK：Sink算子应当忽略-U类型的消息，但是在收到-D类型的消息时删除主键对应的行。​

• CHANGELOG_STANDARD：Sink算子应当在收到-U和-D类型的数据时均会删除主键对应的行。​

合并机制

详情

去重（Deduplicate）

去重机制（deduplicate）是默认的数据合并机制。对于多条具有相同primary key的数据，Paimon结果表仅会保留最新一条数据，并丢弃其它具有primary key的数据。

部分更新（Partial Update）

通过指定部分更新机制（partial-update），您可以通过多条消息对数据进行逐步更新，并最终得到完整的数据。具体来说，具有相同primary key的新数据将会覆盖原来的数据，但值为null的列不会进行覆盖。

例如，假设Paimon结果表按顺序收到了以下三条数据：

• <1, 23.0, 10, NULL>

• <1, NULL, NULL, 'This is a book'>

• <1, 25.2, NULL, NULL>

第一列是primary key，则最终结果为<1, 25.2, 10, 'This is a book'>。

预聚合（Aggregation）

部分场景下，可能只关心聚合后的值。预聚合机制（aggregation）将具有相同primary key的数据根据您指定的聚合函数进行聚合。对于不属于primary key的每一列，都需要通过fields.<field-name>.aggregate-function指定一个聚合函数，否则该列将默认使用last_non_null_value聚合函数。例如，考虑以下Paimon表的定义。

CREATE TABLE MyTable (
  product_id BIGINT,
  price DOUBLE,
  sales BIGINT,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'merge-engine' = 'aggregation',
  'fields.price.aggregate-function' = 'max',
  'fields.sales.aggregate-function' = 'sum'
);

price列将会根据max函数进行聚合，而sales列将会根据sum函数进行聚合。给定两条输入数据 <1, 23.0, 15>和 <1, 30.2, 20>，最终结果为<1, 30.2, 35>。当前支持的聚合函数与对应的数据类型如下：

• sum：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT和DOUBLE。

• min和max：支持DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP和TIMESTAMP_LTZ。

• last_value和last_non_null_value：支持所有数据类型。

• listagg：支持STRING。

• bool_and和bool_or：支持BOOLEAN。