Realtime Compute for Apache Flink:Paimon connector

Item

Description

Supported type

Source table, dimension table, sink table, and data ingestion sink

Running mode

Streaming mode and batch mode

Data format

N/A

Metric

N/A

API type

SQL API and data ingestion YAML API

Data update or deletion in a sink table

Supported

Parameter

Description

Type

Required

Default value

Remarks

connector

The type of the table.

String

No

N/A

• If you create an Apache Paimon table in an Apache Paimon catalog, you do not need to specify this parameter.

• If you create a temporary Apache Paimon table in a catalog of a storage other than Apache Paimon, set the value to paimon.

path

The storage path of the table.

String

No

N/A

• If you create an Apache Paimon table in an Apache Paimon catalog, you do not need to specify this parameter.

• If you create a temporary Apache Paimon table in a catalog of a storage other than Apache Paimon, set this parameter to the HDFS or OSS directory in which you want to store the table.

auto-create

Specifies whether to automatically create an Apache Paimon table file if no Apache Paimon table file exists in the specified path when you create a temporary Apache Paimon table.

Boolean

No

false

Valid values:

• false (default): If no Apache Paimon table file exists in the specified path, an error is returned.

• true: If the specified path does not exist, the system automatically creates an Apache Paimon table file.

bucket

The number of buckets in each partition.

Integer

No

1

Data that is written to the Apache Paimon table is distributed to each bucket based on the columns that are specified by the bucket-key parameter.

bucket-key

The bucket key columns.

String

No

N/A

The columns based on which the data written to the Apache Paimon table is distributed to different buckets.

Separate column names with commas (,). For example, 'bucket-key' = 'order_id,cust_id' indicates that data is distributed to buckets based on the order_id and cust_id columns.

changelog-producer

The incremental data generation mechanism.

String

No

none

Apache Paimon can generate complete incremental data for any input data stream to facilitate downstream data consumption. Each UPDATE_AFTER data record corresponds to an UPDATE_BEFORE data record. Valid values:

• none (default): No incremental data is generated. The downstream consumer can read data from the Apache Paimon table in streaming mode. However, the incremental data that is read by the downstream consumer contains only UPDATE_AFTER data and does not contain UPDATE_BEFORE data.

• input: The input data streams are written to an incremental data file as incremental data in dual-write mode.

• full-compaction: Complete incremental data is generated each time full compaction is performed.

• lookup: Complete incremental data is generated before commit savepoint is performed.

For more information about how to select an incremental data generation mechanism, see the Incremental data generation mechanism section of this topic.

full-compaction.delta-commits

The maximum interval at which full compaction is performed.

Integer

No

N/A

A full compaction is definitely triggered when the number of commit savepoints reaches the value of this parameter.

lookup.cache-max-memory-size

The memory cache size of the Apache Paimon dimension table.

String

No

256 MB

The value of this parameter determines the cache sizes of both the dimension table and the lookup changelog producer.

merge-engine

The mechanism for merging data that has the same primary key.

String

No

deduplicate

Valid values:

• deduplicate: Only the latest data record is retained.

• partial-update: Existing data that has the same primary key as the latest data is overwritten by the latest data in the non-null columns. Data in other columns remains unchanged.

• aggregation: An aggregate function is specified to perform pre-aggregation.

For more information about the data merging mechanism, see the Data merging mechanism section of this topic.

partial-update.ignore-delete

Specifies whether to ignore delete messages.

Boolean

No

false

Valid values:

• true: Delete messages are ignored.

• false: Delete messages are not ignored. You must set how the sink handles delete messages by configuring sequence.field or other parameters. Otherwise, errors like IllegalStateException or IllegalArgumentException occur.

ignore-delete

Specifies whether to ignore delete messages.

Boolean

No

false

Its valid values are the same as partial-update.ignore-delete.

partition.default-name

The default name of the partition.

String

No

__DEFAULT_PARTITION__

If the value of a partition key column is null or an empty string, the value of this parameter is used as the partition name.

partition.expiration-check-interval

The interval at which the system checks partition expiration.

String

No

1h

For more information, see How do I configure automatic partition expiration?.

partition.expiration-time

The validity period of a partition.

String

No

N/A

If the period of time for which a partition exists exceeds the value of this parameter, the partition expires. By default, a partition never expires.

The period for which a partition exists is calculated based on the value of the partition. For more information, see How do I configure automatic partition expiration?.

partition.timestamp-formatter

The pattern that is used to convert a time string into a timestamp.

String

No

N/A

This parameter specifies the pattern that is used to extract the period of time for which a partition exists from the partition value. For more information, see How do I configure automatic partition expiration?.

partition.timestamp-pattern

The pattern that is used to convert a partition value into a time string.

String

No

N/A

This parameter specifies the pattern that is used to extract the period of time for which a partition exists from the partition value. For more information, see the "How do I configure automatic partition expiration?" section of the FAQ about upstream and downstream storage topic.

scan.bounded.watermark

The end condition for bounded streaming mode. If the watermark of data in an Apache Paimon source table exceeds the value of this parameter, the generation of data in the Apache Paimon source table ends.

Long

No

N/A

N/A

scan.mode

The consumer offset of the Apache Paimon source table.

String

No

default

For more information, see How do I specify the consumer offset for an Apache Paimon source table?.

scan.snapshot-id

The ID of the savepoint from which the Apache Paimon source table starts to consume data.

Integer

No

N/A

For more information, see the How do I specify the consumer offset for an Apache Paimon source table?.

scan.timestamp-millis

The point in time from which the Apache Paimon source table starts to consume data.

Integer

No

N/A

For more information, see the How do I specify the consumer offset for an Apache Paimon source table?.

snapshot.num-retained.max

The maximum number of the latest savepoints that can be retained.

Integer

No

2147483647

The savepoint expiration is triggered only if the condition specified by the snapshot.num-retained.max parameter or the snapshot.time-retained parameter is met and the condition specified by the snapshot.num-retained.min parameter is met.

snapshot.num-retained.min

The minimum number of the latest savepoints that can be retained.

Integer

No

10

N/A

snapshot.time-retained

The duration for which savepoints can be retained.

String

No

1h

The savepoint expiration is triggered only if the condition specified by the snapshot.num-retained.max parameter or the snapshot.time-retained parameter is met and the condition specified by the snapshot.num-retained.min parameter is met.

write-mode

The write mode of the Apache Paimon table.

String

No

change-log

Valid values:

• change-log: Data is inserted into, deleted from, and updated in the Apache Paimon table based on the primary key.

• append-only: The Apache Paimon table allows only data insertion and does not support operations based on the primary key. This mode is more efficient than the change-log mode.

For more information about write modes, see Write modes.

scan.infer-parallelism

Specifies whether to automatically infer the degree of parallelism of the Apache Paimon source table.

Boolean

No

false

Valid values:

• true: The parallelism of the Apache Paimon source table is automatically inferred based on the number of buckets.

• false: The default degree of parallelism that is configured based on Ververica Platform (VVP) is used. If the resource configuration is in expert mode, the degree of parallelism that is configured is used.

scan.parallelism

The degree of parallelism of the Apache Paimon source table.

Integer

No

N/A

sink.parallelism

The parallelism of the Apache Paimon sink table.

Integer

No

N/A

sink.clustering.by-columns

The clustering column that is used to write data to the Apache Paimon sink table.

String

No

N/A

For an Apache Paimon append-only table without primary keys, you can specify this parameter in a batch deployment to enable the clustering feature for data writing. After this feature is enabled, data is clustered and displayed in specific columns by size. This improves the query speed of the table

Separate multiple column names with commas (,). Example: 'col1,col2'.

For more information about the clustering feature, see Clustering.

sink.delete-strategy

Specifies the verification strategy for the system to handle retraction (delete and update-before) messages.

Enum

No

NONE

Valid values:

• NONE (default): Verification is not performed.

• IGNORE_DELETE: The sink operator should ignore retraction messages.

• NON_PK_FIELD_TO_NULL: The sink operator should ignore update_before messages; When receiving delete messages, it should retain the corresponding primary key values and delete the values in non-primary-key columns.

  This value is suitable for partial-update scenarios where data is written to a single table from multiple sinks.

• DELETE_ROW_ON_PK: The sink operator should ignore update_before messages and delete data records corresponding to delete messages.

• CHANGELOG_STANDARD: The sink operator should delete the data records corresponding to the received update_before and delete messages.

Data merging mechanism

Description

deduplicate

This is the default value of the merge-engine parameter. If the data merging mechanism is deduplicate and multiple data records have the same primary key, the Apache Paimon sink table retains only the latest data record and discards other data records.

partial-update

The partial-update mechanism allows you to use multiple messages to update data and finally obtain complete data. New data that has the same primary key as the existing data overwrites the existing data. Columns that have NULL values cannot overwrite existing data.

For example, the Apache Paimon sink table receives the following data records in sequence:

• <1, 23.0, 10, NULL>

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

• <1, 25.2, NULL, NULL>

If the first column is the primary key, the final result is <1, 25.2, 10, 'This is a book'>.

aggregation

In specific scenarios, you may require only the aggregated values. The aggregation mechanism can aggregate data that has the same primary key based on the specified aggregate function. You must use fields.<field-name>.aggregate-function to specify an aggregate function for each column that is not the primary key. If you do not specify an aggregate function for a column that is not the primary key, the column uses the last_non_null_value aggregate function by default. The following sample code provides an example on how to create an Apache Paimon table by using the aggregation mechanism:

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'
);

In this example, data in the price column is aggregated based on the max function, and data in the sales column is aggregated based on the sum function. If the input data records are <1, 23.0, 15> and <1, 30.2, 20>, the result is <1, 30.2, 35>. Mappings between the supported aggregate functions and data types:

• sum: DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE

• min and max: DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ

• last_value and last_non_null_value: all data types

• listagg: STRING

• bool_and and bool_or: BOOLEAN