Data Ingestion into Data Lake

更新时间:
复制 MD 格式

The data ingestion into data lake feature of ApsaraMQ for Kafka allows you to automatically and continuously write real-time messages from Kafka Topics into OSS Table Buckets (based on Apache Iceberg format) without deploying additional Flink or Spark jobs, enabling streaming data ingestion into the data lake and building a Lakehouse architecture.

Overview

Data ingestion into data lake is a native data lake ingestion capability provided by ApsaraMQ for Kafka. After this feature is enabled, messages written to a Topic are persisted in the Kafka cluster and simultaneously written to OSS Table Buckets in Iceberg table format, enabling downstream compute engines (MaxCompute, Hologres, Spark, Trino, etc.) to directly query and analyze the data.

You do not need to develop and maintain ETL pipelines to achieve end-to-end integration from real-time messages to a structured data lake.

Scenarios

Real-time data lake construction

Write user clickstreams, IoT sensor data, transaction logs, and other data to OSS tables in real time through Kafka. The data lake always contains the latest data and supports near real-time analytics.

CDC data ingestion into data lake

Capture database changes using tools such as Debezium and send them to Kafka. The data ingestion feature consumes in CDC mode and merges (Upsert) changes into Iceberg tables, achieving near real-time data synchronization.

Low-cost historical data archiving

Kafka retains raw events for a short period while OSS stores structured historical data long-term. Combined with Iceberg's partitioning and compression strategies, this significantly reduces storage costs.

Exactly-once write guarantee

Ingestion progress is maintained in Kafka Leader metadata, completely eliminating dependency on external systems such as KV stores, ensuring strong consistency while eliminating inter-system coupling. This greatly simplifies the overall pipeline logic, prevents data duplication or loss, and meets the requirements of scenarios such as finance and transactions that demand extremely high data accuracy.

Advantages

Advantage

Description

Zero ETL operations

No need to deploy separate Flink/Spark jobs, eliminating the burden of scheduling, monitoring, alerting, and resource management for hundreds of streaming jobs

Unified schema management

Unified validation and evolution based on Kafka Schema Registry, avoiding dirty data and format errors while eliminating metadata fragmentation

Automated data lake maintenance

Automatically handles small file compaction, expired snapshot cleanup, partition optimization, and other Iceberg table maintenance tasks

Cost advantage

Pay-as-you-go based on actual compute resource (CU) usage without pre-provisioning compute clusters, significantly reducing data lake ingestion costs

Open ecosystem

Writes in Iceberg open format, supporting direct queries from MaxCompute, Hologres, Spark, Flink, Trino, Presto, and other engines

How it works

The data flow process of data ingestion into data lake is as follows:

  1. Producers write messages to a Kafka Topic.

  2. The Kafka cluster persists the messages.

  3. The data ingestion module performs data transformation based on the configured parsing mode and write mode.

  4. The transformed data is written to OSS Table Buckets in Iceberg table format.

  5. Downstream compute engines access the table data in OSS through the Iceberg Catalog.

Limits

Item

Description

Supported version

Only the Serverless edition supports data ingestion into data lake. The minimum version requirement is 3.7.0.0

Storage requirement

An OSS Table Bucket must be configured as the Catalog storage

Supported regions

Hangzhou

Billing

Note

The data ingestion into data lake feature is currently in public preview. A notice will be sent one month before official billing begins.

Enable data ingestion into data lake

To use the data ingestion into data lake feature, you must first enable the feature for your instance. After the feature is enabled, data is written to OSS Table Buckets at the Topic level.

Prerequisites

  • An ApsaraMQ for Kafka Serverless instance is created.

Procedure

  1. Log on to the ApsaraMQ for Kafka console.

  2. In the left-side navigation pane, click Instances.

  3. On the Instances page, click the name of the target instance.

  4. In the left-side navigation pane, click Data Ingestion into Data Lake.

  5. On the Data Ingestion into Data Lake page, follow the steps to complete the Enable Data Ingestion into Data Lake configuration. This includes:

    1. Kafka instance version verification. If the version is lower than the required version, you need to upgrade the Kafka instance version.

    2. Navigate to the OSS Table Bucket page to create a Table Bucket, see OSS tables.

    3. Enable Schema Registry.

    4. OSS service authorization.

    5. Catalog association.

    Parameter

    Description

    Catalog configuration

    Add an existing OSS Table Bucket as the data lake Catalog storage

    Service authorization

    Authorize the Kafka service role to have read and write access to the target OSS Table Bucket

  6. After completing the above configuration, click OK to enable the data ingestion into data lake feature.

Result

After successful activation, the Data Ingestion into Data Lake page shows the enabled status, and you can start configuring data ingestion rules for Topics.

Configure Topic data ingestion into data lake

This section describes how to configure data ingestion into data lake for ApsaraMQ for Kafka Topics, writing Topic messages in real time to OSS Table Buckets (Iceberg format).

Prerequisites

  • The data ingestion into data lake feature is enabled. For more information, see Enable data ingestion into data lake.

  • The target Topic is created and has messages being written to it.

Procedure

  1. Log on to the ApsaraMQ for Kafka console.

  2. In the left-side navigation pane, click Instances.

  3. On the Instances page, click the name of the target instance.

  4. In the left-side navigation pane, click Topics.

  5. On the Topics page, click the name of the target Topic.

  6. On the Topic Details page, click Data Ingestion into Data Lake and follow the steps below to complete the configuration.

Step 1: Configure basic properties

Parameter

Description

Required

Topic name

The target Topic for data lake ingestion

Yes

Namespace

The namespace of the Iceberg table, used for logical grouping in the Catalog

Yes

Table name

The Iceberg table name, which defaults to the Topic name and cannot be modified

Yes (auto-filled)

Table partition type

The partitioning strategy of the Iceberg table, used to optimize query performance. See Property description/Table partition types for details.

No

Step 2: Configure write mode

Parameter

Description

Options

Parsing mode

The method of writing messages to Iceberg

Raw archiving, Structured archiving

Write mode

The mode for writing data to the Iceberg table

Append: append mode; CDC: change data capture.

valueConvertType (Value conversion type)

The deserialization method for message Value

raw: raw bytes; string: string; by_schema_id: parse by Schema ID embedded in the message

keyConvertType (Key conversion type)

The deserialization method for message Key

raw: raw bytes; string: string; by_schema_id: parse by Schema ID embedded in the message

TransformType (Transformation type)

The message body structure transformation method

none: no transformation; flatten: flatten nested structures; flatten_debezium: process Debezium CDC events (requires schema-based conversion)

Additional configuration when write mode is CDC:

When CDC write mode is selected, additional configuration is required:

Parameter

Description

Options

CDCType (CDC type)

The processing method for CDC messages

Generic CDC field, Upsert mode

IdColumns (Primary key columns)

The primary key columns of the Iceberg table, used for row-level identification in CDC operations

Custom input

  • Generic CDC field: Requires configuring CDCField.

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

Parameter

Description

Options

CDCField

Determines whether to execute Insert, Update, or Delete operations based on the field in the message that marks the CDC operation type.

Custom input

Step 3: Configure synchronization policy

Parameter

Description

Default value

Commit interval

The commit cycle for writing data to the Iceberg table, in milliseconds (ms), with a value range of 60,000ms to 900,000ms

60,000ms

Error tolerance

The processing strategy when encountering abnormal data

invalid_data

Error tolerance strategy description:

Value

Description

none

Tolerates no errors. The ingestion task stops immediately when encountering abnormal data

invalid_data

Tolerates only invalid data errors. Skips messages with abnormal formats and continues processing

all

Tolerates all errors. Skips all exceptions and continues processing

Complete configuration

After the configuration is complete, click OK to create the Topic data ingestion into data lake configuration.

Result

After the configuration is successfully created, the system automatically begins consuming Topic messages and writing them to the OSS Table Bucket. You can view the ingestion status and operational metrics on the Data Ingestion into Data Lake page, and read data from the Table Bucket.

Monitoring metrics

This section describes the monitoring metrics supported by the data ingestion into data lake feature of ApsaraMQ for Kafka, helping you understand the running status and performance of ingestion tasks.

Topic-level metrics

Metric name

Metric type

Description

Topic to Table Time Lag

Gauge

Represents the gap between the Kafka timestamp of data already written to the target table for this Topic and the current time. It is calculated as: current timestamp − Topic write progress timestamp. The write progress timestamp for each partition is the maximum Kafka record timestamp among all data already written to the target table for that partition; at the Topic level, the minimum value across all active partitions is taken. Therefore, at the Topic level, this is equivalent to the maximum time gap across all active partitions, in milliseconds. This metric reflects how far behind the data in the table is relative to the current time, and is not equivalent to the consumer offset lag. If this value keeps increasing, it may indicate that lakehouse ingestion progress is lagging on certain partitions, or that the source is continuously producing records with inherently old timestamps. To diagnose the root cause, this metric should be evaluated together with offset lag, write throughput, and the distribution of message timestamps.

Topic to Table Offset Lag

Counter

The amount of data in this Topic that has not yet completed ingestion. Calculation: Topic latest data position - current position of data successfully written to the target table. The Topic level sums the lag across all partitions. Reflects the ingestion backlog. It is more low-level than Time Lag, focusing on "how many records have not caught up." If this value keeps increasing, it indicates that ingestion speed is lower than production speed, and attention should be paid to processing capacity, write capability, or resource bottlenecks.

Ingestion Topic consumption throughput

Gauge

The speed at which the ingestion task reads data from the Topic, in Bytes/s. Measures the ingestion task's ability to pull data from the source Topic. If this value is consistently lower than the business write speed to the Topic, backlog will accumulate and Offset Lag will gradually increase.

Ingestion OSS write throughput

Gauge

The speed at which data from this Topic is transformed and written to OSS, in Bytes/s. Measures the ability to actually land data on storage. When this metric is too low or highly volatile, it may indicate bottlenecks in OSS writes, network bandwidth, file commit batches, or local write strategies.

Ingestion Kafka Record processing speed (Message/s)

Gauge

The speed at which data from this Topic successfully advances the ingestion progress, i.e., how many records per second complete processing and are included in the target table progress, in Message/s. Measures overall ingestion processing capability, including reading, parsing, transformation, writing, and progress commit. The higher this value, the more data completes ingestion per unit time; if this value is low while resource usage is high, further investigation of processing logic or write pipeline bottlenecks is usually needed.

Property description

Table partition types

Seven partition types are supported. The following describes the semantics, applicable scenarios, and configuration examples of each type.

Identity (raw value partition)

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

Syntax: [column_name]

Configuration example:

region

Partition effect:

region value

Partition path

cn-hangzhou

region=cn-hangzhou/

us-east-1

region=us-east-1/

eu-west-1

region=eu-west-1/

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

Note: Field cardinality directly equals the number of partitions. High-cardinality fields (such as user IDs) will 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:

create_time value

Partition path

2026-04-27 10:30:00

create_time_year=2026/

2025-12-01 08:00:00

create_time_year=2025/

2024-06-15 12:00:00

create_time_year=2024/

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:

event_time value

Partition path

2026-04-27 10:30:00

event_time_month=2026-04/

2026-03-15 08:00:00

event_time_month=2026-03/

Applicable scenarios: The most common time partitioning method. Suitable for scenarios where logs, event streams, and metric data are 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:

log_time value

Partition path

2026-04-27 10:30:00

log_time_day=2026-04-27/

2026-04-28 08:00:00

log_time_day=2026-04-28/

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:

ingest_time value

Partition path

2026-04-27 10:30:00

ingest_time_hour=2026-04-27-10/

2026-04-27 11:15:00

ingest_time_hour=2026-04-27-11/

Applicable scenarios: Ultra-high throughput scenarios where queries need to be precise to the hour. 8760 partitions are produced 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:

user_id value

Partition path

12345

user_id_bucket=7/ (hash(12345) % 16)

67890

user_id_bucket=2/ (hash(67890) % 16)

11111

user_id_bucket=14/ (hash(11111) % 16)

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:

Data volume (per query range)

Recommended N

< 1GB per day

4~8

1~10GB per day

8~16

10~100GB per day

16~64

> 100GB per day

64~128

Note: N cannot be directly modified 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:

Field type

Truncation rule

Example (W=100)

int / long

value - (value % W)

1234 -> 1200, 567 -> 500, 50 -> 0

string

Take the first W characters

W=3: "hangzhou" -> "han"

decimal

Similar to integer, truncated by precision

-

Configuration example (integer truncation):

truncate(price, 100)

Partition effect:

price value

Partition path

1234

price_trunc=1200/

567

price_trunc=500/

50

price_trunc=0/

99

price_trunc=0/

Configuration example (string truncation):

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

Partition effect:

city_name value

Partition path

hangzhou

city_name_trunc=han/

shanghai

city_name_trunc=sha/

shenzhen

city_name_trunc=she/

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 excessive total partition count after combination (e.g., day x bucket(64) = 64 partitions per day, 23360 per year)

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

Parsing mode description

  • Raw archiving: Stores messages directly in the Iceberg table as raw bytes or strings, suitable for log archiving and raw data backup scenarios. The write mode is fixed to Append and TransformType is fixed to none.

  • Structured archiving: Parses and processes messages based on Schema before writing to the Iceberg table, supporting both Append and CDC write modes, suitable for scenarios requiring structured queries and analysis.

Write mode description

  • Append: Append write mode where all messages are directly appended to the Iceberg table, suitable for append-only data scenarios such as logs and event streams.

  • CDC: Change data capture mode that supports Insert / Update / Delete operations based on the operation type field in messages, suitable for database change synchronization into data lake scenarios.