Paimon append tables-Data Lake Formation(DLF)-阿里云帮助中心

Quick decision guide

Category	Scenario	Recommended configuration
Bucketing mode	Bulk append with no need for order or point lookups (for example, logs, tracking events, ODS detail tables).	Unbucketed (default) `'bucket' = '-1'`
	Streaming reads for the same key must match the write order.	Fixed bucketing `'bucket' = '<num>'`
	Data pruning by key is required, or you want Spark bucketed join optimization.	`'bucket-key' = 'c1,c2'`
Row-level update and delete	Perform lightweight DELETE / UPDATE / MERGE INTO on an Append table via Spark.	`'deletion-vectors.enabled' = 'true'`
Clustering	Queries frequently filter by a specific column, and query performance should be improved.	`'clustering.incremental' = 'true''clustering.columns' = 'c1,c2'`

Note

To store and query multimodal data efficiently, see Multimodal lake.

Basic concepts

A Paimon table created without a primary key is a Paimon Append table. The following SQL creates an Append table partitioned by the dt column:

CREATE TABLE T (
  user_id BIGINT,
  event_time TIMESTAMP(3),
  event_type STRING,
  properties STRING,
  dt STRING
) PARTITIONED BY (dt);

An Append table has the following characteristics:

Append-only writes: Accepts an insert-only data stream. If the upstream is a changelog, ensure it contains no -U/-D records, or set ignore-delete = true (default: false) to ignore -U/-D records.
Full-featured batch read and write: Similar to a Hive partitioned table, with additional capabilities including time travel (version rollback), fast scan planning (based on partition and column statistics plus File Index pruning), schema evolution (adding, dropping, altering, or renaming columns without side effects), and row-level delete and update (via deletion vectors).
Streaming read and write: Supports queue-like streaming reads and writes with minute-level latency.

Compared with a primary-key table, an Append table is suitable for append-oriented data that does not need updates by primary key, such as logs, tracking events, ODS detail tables, binlog landings, and feature or sample tables.

Bucketing modes

Append tables support two bucketing modes.

Unbucketed (default)

An Append table is unbucketed when bucket is not set or is set to 'bucket' = '-1'. Data is written without bucket distinction, and order is not preserved. Advantages of this mode:

Write parallelism is not constrained by the bucket count, allowing higher-concurrency writes.
No need to estimate a bucket count in advance or adjust it manually as data grows.

Suitable for bulk-append scenarios that do not require order or point lookups, such as logs, tracking events, and ODS detail tables.

Fixed bucketing

Set 'bucket' = '<num>' (a positive integer) in the table options and specify bucket keys through 'bucket-key' (multiple keys separated by commas). This defines the bucket count for a non-partitioned table, or the bucket count per partition for a partitioned table.

Advantages over unbucketed mode:

Data within a bucket is read in write order. To read data with the same key in order, set that column as the bucket key.
When query predicates on the bucket key use equality (=) or IN conditions, bucket pruning accelerates the query.
When two Append tables share the same bucket count and their bucket-key is used as the join equality condition, Spark can leverage bucketed join to reduce resource consumption.

Spark bucketed join example:

CREATE TABLE t1 (
  user_id BIGINT,
  event_type STRING
) TBLPROPERTIES (
  'bucket' = '16',
  'bucket-key' = 'user_id'
);

CREATE TABLE t2 (
  user_id BIGINT,
  profile STRING
) TBLPROPERTIES (
  'bucket' = '16',
  'bucket-key' = 'user_id'
);

-- Bucketed Join
SELECT *
FROM t1
JOIN t2
ON t1.user_id = t2.user_id;

Row-level update and delete

An Append table is append-only by default. To run DELETE, UPDATE, or MERGE INTO with Spark, set 'deletion-vectors.enabled' = 'true'. When enabled, delete information is recorded in deletion-vector files, and queries use them to filter out deleted rows. DLF background maintenance jobs periodically merge deletion vectors with data files and reclaim storage.

Clustering

When queries frequently filter by certain columns, enabling clustering reorders data to improve query performance. DLF maintenance jobs perform clustering periodically; no manual execution is required.

Examples:

Flink

-- Unbucketed Append table
CREATE TABLE app_log (
  dt STRING,
  log_time TIMESTAMP,
  user_id BIGINT,
  event_type STRING,
  url STRING
) PARTITIONED BY (dt) WITH (
  'bucket' = '-1',
  'clustering.incremental' = 'true',
  'clustering.columns' = 'user_id,event_type'
);

-- Bucketed Append table
CREATE TABLE app_log (
  dt STRING,
  log_time TIMESTAMP,
  user_id BIGINT,
  event_type STRING,
  url STRING
) PARTITIONED BY (dt) WITH (
  'bucket' = '64',
  'bucket-key' = 'user_id',
  'bucket-append-ordered' = 'false',
  'clustering.incremental' = 'true',
  'clustering.columns' = 'event_type'
);

Spark

-- Unbucketed Append table
CREATE TABLE app_log (
  dt STRING,
  log_time TIMESTAMP,
  user_id BIGINT,
  event_type STRING
) PARTITIONED BY (dt) TBLPROPERTIES (
  'bucket' = '-1',
  'clustering.incremental' = 'true',
  'clustering.columns' = 'user_id,event_type'
);

-- Bucketed Append table
CREATE TABLE app_log (
  dt STRING,
  log_time TIMESTAMP,
  user_id BIGINT,
  event_type STRING
) PARTITIONED BY (dt) TBLPROPERTIES (
  'bucket' = '16',
  'bucket-key' = 'user_id',
  'bucket-append-ordered' = 'false',
  'clustering.incremental' = 'true',
  'clustering.columns' = 'event_type'
);

Important

Clustering reorders files. Because a bucketed Append table is already ordered, you must set 'bucket-append-ordered' = 'false' to enable clustering. Currently, clustering on a bucketed Append table also requires that deletion vectors are not enabled.

Configuration guide

Scenario 1: Queries frequently filter by certain columns and query performance should be improved.

'clustering.incremental' = 'true'
'clustering.columns' = 'event_time,user_id'

Scenario 2: Automatic full clustering after a partition stops receiving writes.

-- A partition with no new writes for three days is treated as a historical partition
'clustering.history-partition.idle-to-full-sort' = '3d'

Resource consumption of storage optimization

The resource consumption of the storage optimization job on an Append table is affected by many factors and is difficult to compute precisely. Empirical value: for a typical table with fewer than 200 columns, each 25 GB of data consumes about 1 CU*H. The actual cost is based on your bill.

Influencing factors:

Number of small files: In an optimization job, opening a file costs roughly the same as reading 4 MB of data. When most files written are small (under 10 MB), the read overhead during compaction is proportionally larger. To avoid too-small files, reduce the bucket count (for bucketed tables), reduce write concurrency (for unbucketed tables), or increase the checkpoint interval.
Shared optimization jobs for small tables: Multiple small tables that meet the criteria can share the same optimization job (up to 64 tables), saving resources. Criteria for a small table:
- data-evolution.enabled = true is not set (default: false).
- clustering.incremental = true is not set (default: false).
- deletion-vectors.enabled = true is not set (default: false).
- Hourly write volume is less than 16 GB.