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:
Producers write messages to a Kafka Topic.
The Kafka cluster persists the messages.
The data ingestion module performs data transformation based on the configured parsing mode and write mode.
The transformed data is written to OSS Table Buckets in Iceberg table format.
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
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
Log on to the ApsaraMQ for Kafka console.
In the left-side navigation pane, click Instances.
On the Instances page, click the name of the target instance.
In the left-side navigation pane, click Data Ingestion into Data Lake.
On the Data Ingestion into Data Lake page, follow the steps to complete the Enable Data Ingestion into Data Lake configuration. This includes:
Kafka instance version verification. If the version is lower than the required version, you need to upgrade the Kafka instance version.
Navigate to the OSS Table Bucket page to create a Table Bucket, see OSS tables.
Enable Schema Registry.
OSS service authorization.
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
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
Log on to the ApsaraMQ for Kafka console.
In the left-side navigation pane, click Instances.
On the Instances page, click the name of the target instance.
In the left-side navigation pane, click Topics.
On the Topics page, click the name of the target Topic.
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:
regionPartition effect:
region value | Partition path |
|
|
|
|
|
|
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 |
|
|
|
|
|
|
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 |
|
|
|
|
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 |
|
|
|
|
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 |
|
|
|
|
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 |
|
|
|
|
|
|
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 |
| 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 |
|
|
|
|
|
|
|
|
Configuration example (string truncation):
# Partition by first 3 characters of city_name
truncate(city_name, 3)Partition effect:
city_name value | Partition path |
|
|
|
|
|
|
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.parquetComposition 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.