Paimon connector - Realtime Compute for Apache Flink - Alibaba Cloud Documentation Center

This topic explains how to use the Paimon connector for a streaming data lakehouse. For best results, we recommend using the connector with the Paimon Catalog.

Background information

Apache Paimon is a unified streaming and batch data lake storage format that supports high-throughput writes and low-latency queries. Paimon integrates well with popular compute engines available in Alibaba Cloud E-MapReduce, such as Flink, Spark, Hive, and Trino. You can use Apache Paimon to quickly build a data lake on HDFS or OSS and connect to these compute engines for data lake analytics. For more information, see Apache Paimon.

Category	Description
Supported types	Source table, dimension table, result table, and target for data ingestion
Running mode	Streaming mode and batch mode
Data format	Not supported
Monitoring metrics	None
API types	SQL and YAML for data ingestion
Updates and deletes on result tables	Yes

Key features

Apache Paimon provides the following core capabilities:

Build a lightweight, low-cost data lake on HDFS or object storage.
Read and write large-scale datasets in both streaming and batch modes.
Run batch and OLAP queries with data freshness from minutes down to seconds.
Ingest and produce incremental data, serving as a storage layer for both traditional offline and modern streaming data warehouses.
Pre-aggregate data to reduce storage costs and downstream compute load.
Access historical versions of data.
Filter data efficiently.
Enable schema evolution.

Limitations and recommendations

The Paimon connector requires Flink compute engine VVR 6.0.6 or later.

The following table lists the version compatibility between Paimon and VVR.

Apache Paimon version	VVR
1.3.1	11.5, 11.6
1.3	11.4
1.2	11.2, 11.3
1.1	11.1
1.0	8.0.11
0.9	8.0.7, 8.0.8, 8.0.9, 8.0.10

Storage recommendations for concurrent writes

When multiple jobs perform concurrent writes to the same Paimon table, using standard OSS storage (oss://) can occasionally cause commit conflicts or job failures due to limitations on atomic file operations.

For stable and consistent writes, use a metadata or storage service that provides strong atomic guarantees. The preferred option is Data Lake Formation (DLF), which offers unified management of Paimon metadata and storage. Alternatively, you can use OSS-HDFS or HDFS.
How configuration changes take effect

Changes to the configuration parameters of a Paimon table take effect only after you restart the related jobs. Running jobs do not dynamically load these changes.
Delayed physical reclamation of dropped partitions

When you run a DROP PARTITION operation, the system does not immediately delete the underlying physical data files.
This operation performs a logical deletion. Paimon removes the metadata of the target partition only from the latest snapshot. Because Paimon supports the time travel feature, historical snapshots still reference that partition's data files. The physical data files are permanently deleted only after all historical snapshots that reference the partition reach their retention limit and are cleaned up by the snapshot expiration mechanism.

SQL

Use the Paimon connector in SQL jobs as a source table or a result table.

Syntax

If you create a Paimon table in a Paimon Catalog, you do not need to specify the connector parameter. The syntax is as follows:
```
CREATE TABLE `<YOUR-PAIMON-CATALOG>`.`<YOUR-DB>`.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
Note
If you have already created a Paimon table in a Paimon Catalog, you can use it directly.
If you create a Paimon temporary table in another Catalog, you must specify the 'connector' and 'path' parameters. The syntax is as follows:
```
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', -- If Paimon table data files do not exist at the specified path, they are automatically created.
  ...
);
```
Note
- Path example: 'path' = 'oss://<bucket>/test/order.db/orders'. Do not omit the .db suffix. Paimon relies on this suffix to identify the database.
- Multiple jobs that write to the same table must share the same path configuration.
- If two path configurations are different, Paimon does not recognize them as the same table. Even if the physical path is the same, inconsistent Catalog configurations can lead to concurrent write conflicts, failed compact operations, and data loss. For example, Paimon considers oss://b/test and oss://b/test/ to be different tables due to the trailing slash, even though they may point to the same physical location.

WITH parameters

Parameter	Description	Type	Required	Default	Remarks
connector	Specifies the connector for the table.	String	No	None	This parameter is not required if you create a Paimon table in a Paimon catalog. If you create a Paimon temporary table in another catalog, this parameter must be set to `paimon`.
path	The table's storage path.	String	No	None	This parameter is not required if you create a Paimon table in a Paimon catalog. If you create a Paimon temporary table in another catalog, this parameter specifies the storage directory of the table in HDFS or OSS.
auto-create	Specifies whether to automatically create table files if they do not exist at the specified path.	Boolean	No	false	Valid values: false (default): If Paimon table files do not exist at the specified path, the job fails. true: If the specified path does not exist, Flink automatically creates the Paimon table files.
file.format	The format of the data files.	String	No	parquet	Valid values: orc parquet avro lance (Supported in Realtime Compute for Apache Flink 11.6 and later)
bucket	The number of buckets per partition.	Integer	No	1	Paimon distributes data to buckets based on the `bucket-key`. Note We recommend that each bucket contain less than 5 GB of data.
bucket-key	The column(s) used as the bucket key.	String	No	None	Specifies the columns used to distribute data into buckets. Separate multiple column names with a comma (,). For example, `'bucket-key' = 'order_id,cust_id'` distributes data based on the `order_id` and `cust_id` columns. Note If this parameter is not specified, Paimon distributes data based on the primary key. If the table has no primary key, Paimon distributes data based on the values of all columns.
changelog-producer	The changelog production mechanism.	String	No	none	Paimon can produce a complete changelog for any input stream, which means that every `update_after` record has a corresponding `update_before` record. This simplifies downstream consumption. Valid values: none (default): No extra changelog is produced. Downstream consumers can still read the Paimon table in streaming mode, but the changelog is incomplete (contains only `update_after` records without corresponding `update_before` records). input: Writes the input stream to changelog files, which serve as the complete changelog. full-compaction: Produces a complete changelog during each full compaction. lookup: Produces a complete changelog before each snapshot is committed. For more information about how to choose a changelog producer, see Changelog Production.
full-compaction.delta-commits	The maximum number of commits between two consecutive full compactions.	Integer	No	None	Specifies the maximum number of snapshot commits to allow before triggering a full compaction.
lookup.cache-max-memory-size	The memory cache size for a Paimon dimension table.	String	No	256 MB	This parameter controls the cache size for both dimension table lookups and the `lookup` changelog producer.
merge-engine	The mechanism for merging records with the same primary key.	String	No	deduplicate	Valid values: deduplicate: Keeps only the latest record. partial-update: Updates the existing record with the non-null values from the latest record. Other columns remain unchanged. aggregation: Performs pre-aggregation by using specified aggregate functions. For a detailed analysis of merge engines, see Merge Engine.
partial-update.ignore-delete	Specifies whether to ignore delete (-D) messages.	Boolean	No	false	Valid values: true: Ignores delete messages. false: Processes delete messages. You must configure a strategy for handling deletions using parameters like sequence.field to prevent potential `IllegalStateException` or `IllegalArgumentException` errors. Note In Realtime Compute for Apache Flink 8.0.6 and earlier, this parameter takes effect only in partial update scenarios where `merge-engine = 'partial-update'`. In Realtime Compute for Apache Flink 8.0.7 and later, this parameter is also compatible with non-partial-update scenarios and has the same function as the `ignore-delete` parameter. We recommend that you use `ignore-delete` instead. Decide whether to enable this parameter based on your business requirements and whether the delete messages are expected. If delete messages do not match the intended semantics of your job, failing fast is often the better choice.
ignore-delete	Specifies whether to ignore delete (-D) messages.	Boolean	No	false	The valid values are the same as those for partial-update.ignore-delete. Note This parameter is supported only in Realtime Compute for Apache Flink 8.0.7 and later. This parameter has the same function as `partial-update.ignore-delete`. We recommend that you use `ignore-delete` and avoid configuring both parameters at the same time.
partition.default-name	The default partition name.	String	No	__DEFAULT_PARTITION__	The partition name to use when a partition column's value is null or an empty string.
partition.expiration-check-interval	How often the system checks for expired partitions.	String	No	1h	For details, see How to configure automatic partition expiration.
partition.expiration-time	The duration after which partitions expire.	String	No	None	A partition expires when its age exceeds this value. By default, partitions never expire. The system calculates a partition's age from its partition value. For details, see How to configure automatic partition expiration.
partition.timestamp-formatter	The format string for converting a time string to a timestamp.	String	No	None	Specifies the format for extracting the partition age from the partition value. For details, see How to configure automatic partition expiration.
partition.timestamp-pattern	The format string for converting a partition value to a time string.	String	No	None	Specifies the pattern for extracting a time string from the partition value. For details, see How to configure automatic partition expiration.
scan.bounded.watermark	The watermark value that signals the end of a scan. The source table stops producing data when its watermark exceeds this value.	Long	No	None	N/A
scan.mode	Specifies the consumption position for the Paimon source table.	String	No	default	For details, see How to set the consumption position for a Paimon source table.
scan.snapshot-id	Specifies the snapshot from which the Paimon source table starts consumption.	Integer	No	None	For details, see How to set the consumption position for a Paimon source table.
scan.timestamp-millis	Specifies the point in time from which the Paimon source table starts consumption.	Integer	No	None	For details, see How to set the consumption position for a Paimon source table.
snapshot.num-retained.max	The maximum number of recent snapshots to retain.	Integer	No	2147483647	Snapshot expiration is triggered if either this condition or the `snapshot.time-retained` condition is met, and the `snapshot.num-retained.min` condition is also met.
snapshot.num-retained.min	The minimum number of recent snapshots to retain.	Integer	No	10	N/A
snapshot.time-retained	The retention period for snapshots.	String	No	1h	Snapshot expiration is triggered if either this condition or the `snapshot.num-retained.max` condition is met, and the `snapshot.num-retained.min` condition is also met.
write-mode	The write mode for the Paimon table.	String	No	change-log	Valid values: change-log: The Paimon table supports insert, delete, and update operations based on the primary key. append-only: The Paimon table accepts only insert operations and does not support a primary key. This mode is more efficient than the change-log mode. For more information about write modes, see Write Mode.
scan.infer-parallelism	Specifies whether to automatically infer the parallelism for a Paimon source table.	Boolean	No	true	Valid values: true: Automatically infers the parallelism of the Paimon source table based on the number of buckets. false: Uses the default parallelism configured in Realtime Compute for Apache Flink. If expert mode is enabled, the job uses your configured parallelism instead.
scan.parallelism	The parallelism for the Paimon source table.	Integer	No	None	Note This parameter is ignored if Resource Mode is set to expert mode on the job's Deployment Details > > Resource Configuration tab.
sink.parallelism	The parallelism for the Paimon sink table.	Integer	No	None	Note This parameter is ignored if Resource Mode is set to expert mode on the job's Deployment Details > > Resource Configuration tab.
sink.clustering.by-columns	Specifies the clustering columns for writing to the Paimon sink table.	String	No	None	For Paimon append-only tables (tables without a primary key), this parameter enables clustered writing in batch jobs. This process improves query performance by grouping data on the specified columns. Separate multiple column names with a comma (,), for example, `'col1,col2'`. For more information about clustering, see the Apache Paimon Official Documentation.
sink.delete-strategy	Specifies a validation strategy to ensure that the system correctly handles retraction messages (-D/-U).	Enum	No	NONE	Valid values and the expected behavior of the sink operator when handling retraction messages: NONE (default): Does not perform validation. IGNORE_DELETE: The sink operator must ignore -U and -D messages. No retraction occurs. NON_PK_FIELD_TO_NULL: The sink operator must ignore -U messages. For -D messages, it keeps the primary key and sets all other non-primary-key fields to null. This is mainly used for partial updates when multiple sinks write to the same table. DELETE_ROW_ON_PK: The sink operator must ignore -U messages but delete the row corresponding to the primary key for -D messages. CHANGELOG_STANDARD: The sink operator must delete the row corresponding to the primary key for both -U and -D messages. Note This parameter is supported only in Realtime Compute for Apache Flink 8.0.8 and later. The actual sink behavior for retractions is determined by other parameters, such as `ignore-delete` and `merge-engine`. This parameter only validates if the actual behavior matches the chosen strategy. If there is a mismatch, the job fails with an error message that suggests corrections.

Note

For more information about configuration options, see the Apache Paimon Official Documentation.

Feature details

Data freshness and consistency

The Paimon sink table uses the two-phase commit protocol to commit data during each Flink job checkpoint. Therefore, data freshness is determined by the Flink job's checkpoint interval. Each commit produces up to two snapshots.

When two Flink jobs write to the same Paimon table concurrently, if the jobs write to different buckets, they achieve serializable consistency. If the jobs write to the same bucket, they only achieve snapshot isolation. This means the table's data might be a mix of results from both jobs, but no data loss occurs.

Merge engine

When a Paimon sink table receives multiple records with the same primary key, it merges them into a single record to maintain uniqueness. You can control this behavior by setting the merge-engine parameter. The following table describes the available merge engines.

Merge engine	Description
Deduplicate	The deduplication engine is the default. For multiple records with the same primary key, the Paimon sink table keeps only the latest record and discards the others. Note If the latest record is a delete message, all records with that primary key are discarded.
Partial Update	The partial update engine allows you to build a complete record by incrementally updating it with multiple messages. When a new record with the same primary key arrives, its non-null values overwrite the corresponding fields in the existing record. The engine ignores fields that are null in the new record and retains the existing values. For example, assume a Paimon sink table receives the following three records in order: <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 merged record is <1, 25.2, 10, 'This is a book'>. Note To stream-read the results of a partial update, you must set the `changelog-producer` parameter to `lookup` or `full-compaction`. The partial update engine cannot process delete messages. You can set the `partial-update.ignore-delete` parameter to `true` to ignore delete messages.
Aggregation	In some use cases, you might only need the aggregated value of records. The aggregation engine combines records that have the same primary key by using the aggregate functions you specify. For each non-primary key column, you must specify an aggregate function using the `fields.<field-name>.aggregate-function` option. Otherwise, the column defaults to using the `last_non_null_value` aggregate function. For example, consider the following Paimon table definition. `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' );` The `price` column is aggregated using the `max` function, and the `sales` column is aggregated using the `sum` function. Given two input records, <1, 23.0, 15> and <1, 30.2, 20>, the final result is <1, 30.2, 35>. The supported aggregate functions and their corresponding data types are: sum: Supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, and DOUBLE. min and max: Supports DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ. last_value and last_non_null_value: Supports all data types. listagg: Supports STRING. bool_and and bool_or: Supports BOOLEAN. Note Only the `sum` function supports retractions and deletions; other aggregate functions do not. If you need certain columns to ignore retraction and delete messages, you can set `'fields.${field_name}.ignore-retract'='true'`. To stream-read the results of an aggregation, you must set the `changelog-producer` parameter to `lookup` or `full-compaction`.

Changelog producer

Set the changelog-producer parameter to configure Paimon to generate a complete changelog (where every update_after record has a corresponding update_before record) for any input stream. The following table describes the available changelog producers. For more details, see the Apache Paimon official documentation.

Producer	Description
None	When you set `changelog-producer` to `none` (the default), downstream Paimon source tables see only the latest state of the data for a given primary key. This incomplete changelog makes it difficult for consumers to perform correct calculations because they cannot determine the previous state of the data, only whether it was deleted or what its latest state is. For example, if a downstream consumer needs to calculate a column's sum and only sees the latest value of 5, it cannot determine how to update the total. If the previous value was 4, the sum should increase by 1; if the previous value was 6, the sum should decrease by 1. Consumers that are sensitive to `update_before` records should not use the `none` producer, but other changelog producers introduce performance overhead. Note If your downstream consumer, such as a database, is not sensitive to `update_before` data, you can use the `none` producer. Configure the changelog producer based on your specific requirements.
Input	When you set `changelog-producer` to `input`, the sink table dual-writes the input stream to changelog files. Use this producer only when the input stream itself is already a complete changelog, such as data from Change Data Capture (CDC).
Lookup	When you set `changelog-producer` to `lookup`, the sink table uses a point-query mechanism, similar to a dimension table lookup, to generate a complete changelog for the current snapshot before it is committed. This producer generates a complete changelog from any input stream. Compared to the `full-compaction` producer, the `lookup` producer offers better timeliness for the changelog but consumes more resources overall. Use this option for use cases that require high data freshness (for example, minute-level).
Full Compaction	When you set `changelog-producer` to `full-compaction`, the sink table generates a complete changelog during each full compaction. This producer generates a complete changelog from any input stream. The interval for full compaction is specified by the `full-compaction.delta-commits` parameter. Compared to the `lookup` producer, the `full-compaction` producer has higher latency but leverages the existing full compaction process without adding extra computation. This results in lower overall resource consumption. Use this option for use cases with low data freshness requirements (for example, hourly).

Write mode

Paimon tables support the following write modes.

Mode

Description

Change-log

The change-log write mode is the default for Paimon tables. This mode supports insert, delete, and update operations based on a primary key. You can also use the merge engines and changelog producers in this mode.

Append-only

The append-only write mode supports only data insertions and does not use a primary key. This mode is more efficient than the change-log mode and can be used as an alternative to a message queue in scenarios with moderate data freshness requirements (for example, minute-level freshness).

For a detailed description of the append-only write mode, see the Apache Paimon official documentation. When using this mode, note the following:

Set the bucket-key parameter according to your requirements. Otherwise, the Paimon table buckets data based on the values of all columns, which is computationally inefficient.
The append-only write mode can guarantee the output order of data to some extent. The specific output order is as follows:
1. For records from different partitions: if the scan.plan-sort-partition parameter is set, the record from the partition with the smaller value is output first. Otherwise, the record from the earlier-created partition is output first.
2. For records from the same partition and bucket, the earlier-written record is output first.
3. For records from the same partition but different buckets, the output order is not guaranteed because different buckets are processed by different concurrent tasks.

Target for CTAS and CDAS

Paimon tables support real-time data synchronization for single tables or entire databases. Schema changes in upstream tables are also synchronized to Paimon tables in real time. For details, see Manage Paimon tables and Manage Paimon Catalog.

Variant read pruning

Reads only the Variant fields referenced by the query, reducing I/O and memory overhead.

How to enable

Parameter	Default value	Description
`variant.read.pushdown.enabled`	false	Set to `true` to enable Variant read pruning.

Prerequisites

The Variant data must be written by the current version. Variant data written by earlier versions or other engines does not support read pruning.

Supported scenarios

Read pruning takes effect when SQL accesses Variant fields by string key, for example:

SELECT v['a']            FROM t;   -- Single level
SELECT v['a']['b']       FROM t;   -- Nested
SELECT v['a'], v['b']    FROM t;   -- Multiple fields
SELECT id, v['a']        FROM t;   -- Mixed with regular columns
SELECT v['a'] + 1        FROM t;   -- Referenced in expressions

Unsupported scenarios

Referencing the entire Variant directly, such as SELECT v FROM t.
Combining direct reference with field access, such as SELECT v, v['a'] FROM t.
Array index access, such as v[0].
The table schema contains nested Row fields. In this case, read pruning does not apply to any Variant field in the table.

Use Paimon as a dimension table

Paimon tables can be used as dimension tables. For the JOIN syntax, see Dimension table JOIN statements.

By default, lookup loads all data in each parallel instance. This approach is suitable only for small dimension tables. For large dimension tables, use the Shuffle Lookup solutions described below.

Partitioned dimension tables

If your dimension table is partitioned and you only need data from the latest one or two partitions, you can use the dynamic partition loading feature:

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

Parameter	Data type	Default value	Description
lookup.dynamic-partition	String	N/A	`max_pt()`: loads only the latest partition. `max_two_pt()`: loads only the latest two partitions.
lookup.dynamic-partition.refresh-interval	Duration	1 h	The interval at which the system checks for partition updates in the dimension table.

Large dimension tables: fixed-bucket tables

Supported only in VVR 8.0.8 and later. For fixed-bucket tables (bucket > 0), you can use Shuffle Lookup to distribute data by bucket key across parallel instances, so each instance loads only the data in its assigned buckets:

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Note

The join key must be the bucket key. The bucket key defaults to the primary key.
Only fixed-bucket tables (bucket > 0) support this feature.

Large dimension tables: non-fixed-bucket tables

Supported only in VVR 8.0.10 and later. For dynamic-bucket tables or append tables, you can use SHUFFLE_HASH or REPLICATED_SHUFFLE_HASH so that each parallel instance reads all data but retains only the portion it needs:

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

For more information about SHUFFLE_HASH and REPLICATED_SHUFFLE_HASH, see Dimension table JOIN statements.

Data ingestion

You can use the Paimon connector as a sink in data ingestion YAML jobs.

Syntax

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

Parameters

Parameter	Description	Required	Type	Default	Notes
type	The connector type.	Yes	STRING	None	The value must be `paimon`.
name	The name of the sink.	No	STRING	None
catalog.properties.metastore	The type of the Paimon catalog.	No	STRING	filesystem	Valid values: filesystem (default) rest (supports only Data Lake Formation (DLF), not DLF-Legacy)
catalog.properties.*	Parameters for creating a Paimon catalog.	No	STRING	None	For more information, see Manage Paimon Catalog.
table.properties.*	Parameters for creating a Paimon table.	No	STRING	None	For more information, see Paimon table options.
catalog.properties.warehouse	The root directory for file storage.	No	STRING	None	This parameter applies only when `catalog.properties.metastore` is set to `filesystem`.
commit.user-prefix	The username prefix for committing data files.	No	STRING	None	Note We recommend setting different usernames for different jobs. This makes it easier to identify the job that causes a commit conflict.
partition.key	The partition keys for a partitioned table.	No	STRING	None	Different tables are separated by `;`, different fields are separated by `,`, and tables and fields are separated by `:`. For example: `testdb.table1:id1,id2;testdb.table2:name`.
sink.cross-partition-upsert.tables	Lists tables that require a cross-partition upsert, where the primary key does not include all partition keys.	No	STRING	None	Format: Use a semicolon `;` to separate table names. Performance recommendation: This operation is resource-intensive. Create separate jobs for these tables. Important You must list all tables that meet the criteria. Omitting a table name will result in data duplication.
sink.commit.parallelism	Specifies the parallelism of the Commit operator.	No	INTEGER	None	If the Commit operator is a bottleneck, use this parameter to increase its parallelism and improve performance. This parameter is supported only in Realtime Compute for Apache Flink 11.6 and later. Note Setting this parameter changes the operator parallelism. When restarting a stateful job, you must specify `AllowNonRestoredState` so the job can ignore partial operator state.

Reuse an existing catalog

Starting with Realtime Compute for Apache Flink 11.5, you can directly reference a built-in Paimon catalog from the Data Management page in a Flink CDC data ingestion job. This reduces manual configuration.

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

Data ingestion jobs can automatically reuse all parameters in a Paimon catalog. This is equivalent to manually configuring parameters prefixed with catalog.properties. in a YAML job.

To override an automatically reused parameter, explicitly set it in the YAML job. The explicit YAML configuration has a higher priority. For example, in the preceding sample, the fs.oss.endpoint parameter uses the value from the YAML job, overriding the one in paimon_dlf_catalog.

Examples

When using Paimon as a data ingestion sink, refer to the following examples to configure the job based on your Paimon catalog type.

Example configuration for writing to Object Storage Service (OSS) with a filesystem Paimon catalog:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx

For information about the parameters prefixed with catalog.properties, see Create a Paimon Filesystem Catalog.

Example configuration for writing to Data Lake Formation (DLF) with a rest Paimon catalog:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

For information about the parameters prefixed with catalog.properties, see Flink CDC Catalog Configuration Parameters.

Schema Changes

When used as a data ingestion sink, Paimon supports the following schema change events:

CREATE TABLE EVENT
ADD COLUMN EVENT
ALTER COLUMN TYPE EVENT (Changing the data type of a primary key column is not supported.)
RENAME COLUMN EVENT
DROP COLUMN EVENT
TRUNCATE TABLE EVENT
DROP TABLE EVENT

Note

If the downstream Paimon table already exists, the job writes to the existing schema and does not attempt to create the table again.

Realtime Compute for Apache Flink:Paimon connector

Background information

Key features

Limitations and recommendations

SQL

Syntax

WITH parameters

Feature details

Data freshness and consistency

Merge engine

Changelog producer

Write mode

Target for CTAS and CDAS

Variant read pruning

Use Paimon as a dimension table

Partitioned dimension tables

Large dimension tables: fixed-bucket tables

Large dimension tables: non-fixed-bucket tables

Data ingestion

Syntax

Parameters

Reuse an existing catalog

Examples

Schema Changes

FAQ