ApsaraMQ for Kafka:Data Ingestion into Data Lake

Step 1: Configure basic properties

Step 2: Configure write mode

Additional configuration when write mode is CDC:

When CDC write mode is selected, you must configure the following additional parameters:

• Generic CDC field: Requires configuring CDCField.

• Upsert mode: Performs Upsert (update if exists, insert if not) operations based on primary key columns.

Step 3: Configure synchronization policy

Error tolerance strategy description:

Identity (raw value partition)

Partitions directly by the raw field value. Specify the field name in the configuration without any transform function.

Syntax: [column_name]

Configuration example:

region

Partition effect:

Applicable scenarios: Low-cardinality discrete fields such as region, environment, and business line.

Note: Field cardinality directly determines the number of partitions. High-cardinality fields (such as user IDs) cause partition explosion and should use bucket instead.

year (partition by year)

Extracts the year part of a timestamp/date type field.

Syntax: [year(column)]

Configuration example:

year(create_time)

Partition effect:

Applicable scenarios: Long-cycle data with large time spans that are archived or queried by year (such as historical orders, audit logs).

month (partition by month)

Extracts the year-month part of a timestamp/date type field.

Syntax: [month(column)]

Configuration example:

month(event_time)

Partition effect:

Applicable scenarios: The most common time partitioning method. Suitable for logs, event streams, and metric data queried by month. Only 12 partitions are produced per year, keeping metadata growth under control.

day (partition by day)

Extracts the year-month-day part of a timestamp/date type field.

Syntax: [day(column)]

Configuration example:

day(log_time)

Partition effect:

Applicable scenarios: High-volume log analysis and operational reports queried by day. 365 partitions are produced per year, requiring snapshot expiration to clean up historical partitions.

hour (partition by hour)

Extracts the year-month-day-hour part of a timestamp type field.

Syntax: [hour(column)]

Configuration example:

hour(ingest_time)

Partition effect:

Applicable scenarios: Ultra-high throughput scenarios where queries need to be precise to the hour. This produces 8,760 partitions per year with fast metadata growth. Note:

• Data volume per partition per hour should be sufficiently large (at least tens of MB); otherwise small file problems become severe

• Recommend using RewriteDataFiles for periodic compaction

bucket (hash bucket partition)

Applies Murmur3 hash to field values and distributes them across N fixed buckets. Suitable for uniform distribution of high-cardinality fields.

Syntax: [bucket(column, N)], where N is the number of buckets (positive integer)

Configuration example:

# Hash user_id into 16 buckets
bucket(user_id, 16)

Partition effect:

Applicable scenarios: High-cardinality fields such as user IDs, order IDs, and device IDs where data needs to be distributed but identity cannot be used directly.

Recommended bucket count N:

Note: N cannot be modified directly once set. Modifying the configuration triggers Partition Evolution, where old data remains under the old bucket count and new data uses the new bucket count.

truncate (truncation partition)

Truncates field values by width W, mapping continuous values to discrete partitions.

Syntax: [truncate(column, W)], where W is the truncation width (positive integer)

Truncation behavior for different data types:

Configuration example (integer truncation):

truncate(price, 100)

Partition effect:

Configuration example (string truncation):

# Partition by first 3 characters of city_name
truncate(city_name, 3)

Partition effect:

Applicable scenarios: Numeric range queries (price ranges, age groups, score ranges), string prefix aggregation (city names, product code prefixes).

Composite partitions

Multiple partition fields can be combined using comma separation. Iceberg constructs multi-level partitions in the declared order:

# Time + hash bucket (most common combination)
[day(event_time), bucket(user_id, 16)]

# Region + monthly
[region, month(create_time)]

# Three-level partition
[year(ts), bucket(tenant_id, 8), region]

Directory structure example for composite partitions (time + hash bucket):

data/${hash}/
  event_time_day=2026-04-27/
    user_id_bucket=0/
      00001.parquet
    user_id_bucket=1/
      00002.parquet
  event_time_day=2026-04-28/
    user_id_bucket=0/
      00003.parquet

Composition principles:

• Place the most commonly used query filter conditions first (benefits partition pruning)

• Avoid an excessive total partition count after combination (for example, day x bucket(64) = 64 partitions per day, or 23,360 per year)

• Use identity for low-cardinality fields, bucket for high-cardinality fields