Paimon连接器_实时计算 Flink版(Flink)-阿里云帮助中心

流式数据湖仓Paimon连接器推荐配合Paimon Catalog使用，本文为您介绍如何使用流式数据湖仓Paimon连接器。

背景信息

Apache Paimon是一种流批统一的湖存储格式，支持高吞吐的写入和低延迟的查询。目前阿里云开源大数据平台E-MapReduce常见的计算引擎（例如Flink、Spark、Hive或Trino）都与Paimon有着较为完善的集成度。您可以借助Apache Paimon快速地在HDFS或者云端OSS上构建自己的数据湖存储服务，并接入上述计算引擎实现数据湖的分析，详情请参见Apache Paimon。

类别	详情
支持类型	源表、维表和结果表
运行模式	流模式和批模式
数据格式	暂不支持
特有监控指标	暂无
API种类	SQL
是否支持更新或删除结果表数据	是

特色功能

目前Apache Paimon提供以下核心能力：

基于HDFS或者对象存储构建低成本的轻量级数据湖存储服务。
支持在流模式与批模式下读写大规模数据集。
支持分钟级到秒级数据新鲜度的批查询和OLAP查询。
支持消费与产生增量数据，可作为传统的离线数仓和新型的流式数仓的各级存储。
支持预聚合数据，降低存储成本与下游计算压力。
支持回溯历史版本的数据。
支持高效的数据过滤。
支持表结构变更。

使用限制

仅Flink计算引擎VVR 6.0.6及以上版本支持Paimon连接器。

语法结构

如果您在Paimon Catalog中创建Paimon表，则无需指定connector参数，此时创建Paimon表的语法结构如下。
```
CREATE TABLE `<your-paimon-catalog>`.`<your-db>`.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
说明
如果您已在Paimon Catalog中创建了Paimon表，后续无需再次创建表即可直接使用。

如果您在其他Catalog中创建Paimon临时表，则需要指定connector参数与Paimon表的存储路径path，此时创建Paimon表的语法结构如下。

CREATE TEMPORARY TABLE paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'paimon',
  'path' = '<path-to-paimon-table-files>',
  'auto-create' = 'true', -- 如果指定路径不存在Paimon表数据文件，则会自动创建文件。
  ...
);

WITH参数

参数	说明	数据类型	是否必填	默认值	备注
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的数据量在5 GB以下。
bucket-key	分桶关键列。	String	否	无	指定将写入Paimon表的数据按哪些列的值打散至不同的Bucket中。列名之间用英文逗号（,）分隔，例如`'bucket-key' = 'order_id,cust_id'`会将数据按order_id列和cust_id列的值进行打散。说明如果该参数未填写，则按primary key进行打散。如果Paimon表未指定primary key，则按所有列的值进行打散。
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	在partial-update合并机制中，是否忽略delete消息。	Boolean	否	false	参数取值如下： true：忽略delete消息。 false：碰到delete消息，Flink系统会报错。说明并非所有场景下忽略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	否	无	无。

说明

更多配置项详情请参见Apache Paimon官方文档。

特色功能详解

数据新鲜度与一致性保证

Paimon结果表使用两阶段提交协议，在每次Flink作业的checkpoint期间提交写入的数据，因此数据新鲜度即为Flink作业的checkpoint间隔。每次提交将会产生至多两个snapshot。

当两个Flink作业同时写入一张Paimon表时，如果两个作业的数据没有写入同一个分桶，则能保证serializable级别的一致性。如果两个作业的数据写入了同一个分桶，则只能保证snapshot isolation级别的一致性。也就是说，表中的数据可能混合了两个作业的结果，但不会有数据丢失。

数据合并机制

当Paimon结果表收到多条具有相同primary key的数据时，为了保持primary key的唯一性，Paimon结果表会将这些数据合并成一条数据。通过指定merge-engine参数，您可以指定数据合并的具体行为。数据合并机制详情如下表所示。

合并机制	详情
去重（Deduplicate）	去重机制（deduplicate）是默认的数据合并机制。对于多条具有相同primary key的数据，Paimon结果表仅会保留最新一条数据，并丢弃其它具有primary key的数据。说明如果最新一条数据是一条delete消息，所有具有该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'>。说明如果需要流读partial-update的结果，必须将`changelog-producer`参数设置为lookup或full-compaction。 partial-update无法处理delete消息。您可以设置`partial-update.ignore-delete`参数以忽略delete消息。
预聚合（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。说明只有sum函数支持回撤与删除数据，其它聚合函数不支持回撤与删除。如果您需要某些列忽略回撤与删除消息，可以设置`'fields.${field_name}.ignore-retract'='true'`。如果需要流读aggregation的结果，必须将`changelog-producer`参数设置为lookup或full-compaction。

增量数据产生机制

通过changelog-producer参数设置相应的增量数据产生机制，Paimon可以为任意输入数据流产生完整的增量数据（所有的update_after数据都有对应的update_before数据）。以下列举了所有的增量数据产生机制，更加详细的介绍请参见Apache Paimon官方文档。

机制	详情
None	设置`changelog-producer`为none（默认值）后，此时，对于同一个primary key，下游的Paimon源表只能看到数据的最新情况。但这些最新情况无法让下游消费者方便地了解完整的增量数据，从而进行正确的计算。因为它只能确定对应数据是否被删除了，或最新数据是什么，无法得知更改之前的数据是什么。例如，假设下游消费者需要计算某一列的总和，如果消费者只看到了最新数据5，它无法断定该如何更新总和。因为如果之前的数据是4，它应该将总和增加1；如果之前的数据是6，它应该将总和减去1。此类消费者对update_before较为敏感，建议不要将增量数据产生机制配置为None，但是其他增量数据产生机制会带来性能损耗。说明如果您的下游是数据库之类的对update_before数据不敏感的消费者，则可以将增量数据产生机制配置为None。因此，建议您根据实际需要配置增量数据产生机制。
Input	设置`changelog-producer`为input后，Paimon结果表会将输入数据流双写至增量数据文件中，作为增量数据。因此，只有当输入数据流本身是完整的增量数据时（例如CDC数据），才能使用这一增量数据产生机制。
Lookup	设置`changelog-producer`为lookup后，Paimon结果表会通过一种类似于维表的点查机制，在每次commit snapshot之前产生本次snapshot对应的完整增量数据。无论输入数据是否为完整的增量数据，这一增量数据产生机制均能产生完整的增量数据。与下文的Full Compaction机制相比，Lookup机制产生增量数据的时效性更好，但总体来看耗费的资源更多。推荐在对增量数据的新鲜度有较高要求（例如分钟级）的情况下使用。
Full Compaction	设置`changelog-producer`为full-compaction后，Paimon结果表会在每一次full compaction时产生完整的增量数据。无论输入数据是否为完整的增量数据，这一增量数据产生机制均能产生完整的增量数据。Full compaction的时间间隔由`full-compaction.delta-commits`参数指定。与上文的Lookup机制相比，Full Compaction机制产生增量数据的时效性更差，但它利用了数据的full compaction过程，不产生额外计算，因此总体来看耗费的资源更少。推荐在对增量数据的新鲜度要求不高（例如小时级）的情况下使用。

写入模式

Paimon表目前支持的写入模式如下。

模式

详情

Change-log

change-log写入模式是Paimon表的默认写入模式。该写入模式支持根据primary key对数据进行插入、删除与更新，您也可以在该写入模式下使用上文提到的数据合并机制与增量数据产生机制。

Append-only

append-only写入模式仅支持数据的插入，且不支持primary key。该模式比change-log模式更加高效，可在对数据新鲜度要求一般的场景下（例如分钟级新鲜度）作为消息队列的替代品。

关于append-only写入模式的详细介绍，请参见Apache Paimon官方文档。在使用append-only写入模式时，需要注意以下两点：

建议您根据实际需求设置bucket-key参数，否则Paimon表将根据所有列的值进行分桶，计算效率较低。
append-only写入模式可在一定程度上保证数据的产出顺序，具体的产出顺序为：
1. 如果两条数据来自不同的分区，若设置了scan.plan-sort-partition参数，则分区值较小的数据将首先产出。否则来自较早创建的分区的数据将首先产出。
2. 如果两条数据来自同一分区的同一分桶，则较早写入的数据将首先产出。
3. 如果两条数据来自同一分区的不同分桶，由于不同分桶由不同的并发进行处理，因此不保证两条数据的产出顺序。

作为CTAS和CDAS的目标端

Paimon表支持实时同步单表或整库级别的数据，在同步过程之中如果上游的表结构发生了变更也会实时同步到Paimon表中。详见作为CTAS的目标端Catalog和作为CDAS的目标端Catalog。