MaxCompute:Near-real-time incremental and full data integration architecture with Delta Tables

Table creation

MaxCompute introduces Delta tables with a unified table data format to support the integrated architecture. Delta tables support all features of existing batch processing workflows and new workflows such as near real-time incremental data storage and processing.

To create a Delta table, specify primary keys and set "transactional"="true" in the CREATE TABLE statement:

CREATE TABLE tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING) tblproperties ("transactional"="true");
CREATE TABLE par_tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING)  PARTITIONED BY (pt STRING) tblproperties ("transactional"="true");

Primary keys ensure row uniqueness. The transactional property enables the ACID (atomicity, consistency, isolation, and durability) transaction mechanism with snapshot isolation for read and write operations. For more details, see Table operations.

Key parameters for Delta tables

For the full parameter reference, see the "Parameters for Delta tables" section in Table operations.

write.bucket.num

Specifies the number of buckets per partitioned or non-partitioned table, and the number of concurrent write nodes. The default value is 16. Valid values: (0, 4096].

• For partitioned tables: the value can be changed and automatically applies to new partitions.

• For non-partitioned tables: the value cannot be changed after creation.

acid.data.retain.hours

Specifies the time range of historical data available for time travel queries. The default value is 24. Valid values: [0, 168] (unit: hours).

• Set to 0 to disable time travel queries and significantly reduce historical data storage costs.

• For historical data older than 168 hours (7 days), contact MaxCompute technical support.

Schema evolution

Delta tables support complete schema evolution, including adding and deleting columns. When querying historical data with time travel, the system reads data based on the schema at that historical version.

The following example adds a column:

ALTER TABLE tt2 ADD columns (val2 string);

For the full DDL syntax, see Table operations.

Table data formats

The figure above shows the data structure of a partitioned table. Data files are physically isolated by partition (stored in separate directories), and each partition's data is split into buckets. Delta tables use two types of data files:

Automatic data governance and optimization

Problem: small-file bloat

Delta tables support near real-time incremental data import in minutes. In high-traffic write scenarios with many buckets, the number of small incremental data files can grow quickly, leading to excessive access requests, high costs, and low I/O efficiency. Heavy UPDATE and DELETE operations further compound this by generating large numbers of redundant intermediate historical records.

Solution: four automated governance services

The MaxCompute storage engine automatically governs and optimizes stored data without manual configuration. The storage engine intelligently identifies data characteristics across multiple dimensions and applies policies automatically.

Auto partial compact only clears historical records whose creation time is beyond the time travel retention window.

For scenarios requiring peak query performance, trigger a major compaction manually:

SET odps.merge.task.mode=service;
ALTER TABLE tt2 compact major;

For more information, see COMPACTION.

Near real-time upserts in minutes

Why Delta tables: Traditional batch processing imports incremental data to a new table or partition over hours or days, then triggers an offline ETL process to join and merge that incremental data with existing table data. This has long latency and high resource and storage costs.

With Delta tables, the upsert pipeline maintains a latency of 5 to 10 minutes from data write to query. No complex ETL merge process is required, which reduces both compute and storage costs.

A variety of data sources are common in production—databases, log systems, message queues. MaxCompute provides an open-source Flink connector plug-in that works with DataWorks Data Integration and other data import tools. It supports custom design and development optimization for high concurrency, fault tolerance, and transaction submission scenarios.

Key capabilities of the Flink connector integration:

Traffic throughput varies by environment and configuration. Estimate maximum throughput based on the processing capacity of a single bucket (1 MB/s). The shared Tunnel resource group is used by default for MaxCompute Tunnel, which may cause unstable throughput under heavy resource contention. Limits are also imposed on resource consumption.

Real-time data synchronization from databases using DataWorks Data Integration

Many production systems combine online transaction processing (OLTP), online analytical processing (OLAP), and offline analysis engines. A common workflow is to synchronize new records from a single table or an entire database to MaxCompute in real time for analysis.

The figure above contrasts two approaches:

• Left (batch processing): Incremental data is imported to a new table or partition over hours or days. An offline ETL process then joins and merges the incremental data with existing table data. This has high latency and high resource and storage costs.

• Right (integrated architecture): New records are read from databases in minutes. No periodic data extraction or merging is needed—Delta tables handle updates directly, minimizing compute and storage costs.

Batch processing using SQL DML statements and upserts

The SQL engine's Compiler, Optimizer, and Runtime modules are modified and optimized for Delta table operations. This includes syntax parsing, optimization plans, primary key-based deduplication logic, and runtime upserts to provide full SQL syntax support.

Key behaviors:

• Transaction consistency: After data processing completes, the metadata service performs transaction conflict detection and atomic metadata updates, ensuring read/write isolation and transaction consistency.

• Simplified upserts: The system automatically merges records based on primary keys during queries on Delta tables. For scenarios that mix INSERT and UPDATE operations, use INSERT INTO instead of complex UPDATE or MERGE INTO syntax—this reduces read I/O and saves compute resources.

For the full SQL DML syntax, see DML operations.

Time travel queries

Time travel queries let you query historical versions of a Delta table. Common use cases include:

• Data recovery: Restore data to a specified historical version after accidental modification or deletion.

• Historical backtracking: Audit or reanalyze business data from a past point in time.

Example queries:

-- Query historical data at a specific timestamp.
SELECT * FROM tt2 TIMESTAMP AS OF '2024-04-01 01:00:00';
-- Query historical data from 5 minutes before the current time.
SELECT * FROM tt2 TIMESTAMP AS OF CURRENT_TIMESTAMP() - 300;
-- Query historical data from the second-to-last commit.
SELECT * FROM tt2 TIMESTAMP AS OF GET_LATEST_TIMESTAMP('tt2', 2);

The following figure shows how a time travel query works:

The example uses a transactional table named src:

• Left side (data update process): Transactions t1 through t5 each generate a delta data file. COMPACTION runs at t2 and t4, producing compacted files c1 and c2. In c1, the intermediate historical record (2,a) is deleted and the latest record (2,b) is retained.

• Query resolution: To query historical data at t1, the system reads only delta file d1. To query at t2, it reads compacted file c1 and returns three records. To query at t3, it reads c1 and delta file d3, merging them for output. More frequent COMPACTION accelerates queries but increases operational overhead—choose a trigger policy based on your requirements.

The SQL syntax supports constants, common functions, and the TIMESTAMP AS OF expr and VERSION AS OF expr clauses for precise historical queries. For details, see Time travel queries.

Incremental queries

MaxCompute designs and develops a new SQL incremental query syntax to optimize incremental queries and incremental computing for Delta tables. After you submit a SQL incremental query statement, the MaxCompute engine parses the historical incremental data versions to query, retrieves the relevant compacted data files, merges the file data, and returns the output.

The following figure shows the incremental query process:

The example uses the same transactional table src with transactions t1 through t5 and compacted files c1 (at t2) and c2 (at t4):

• If begin is t1-1 and end is t1, the system reads only delta file d1 at t1.

• If end is t2, the system reads delta files d1 and d2.

• If begin is t1 and end is t2-1, the query range spans t1 to t2. No incremental data exists in this range, so empty rows are returned.

Data in compacted files c1 and c2 generated by COMPACTION is not considered new data for incremental query output.

For the incremental query syntax and parameter limits, see the "Parameters and limits of incremental queries" section in Time travel queries and incremental queries.

Optimized primary key-based data skipping

Delta table data distribution and indexes are built on primary key column values. When you query by primary key, the system filters at multiple levels to significantly reduce the data read—improving query efficiency by hundreds to thousands of times.

Example: A Delta table contains 100 million records. Filtering by a single primary key value may require reading only 10,000 records.

The three-level filtering process:

1. Bucket pruning: Locates the bucket containing the target primary key, eliminating scans of all other buckets.

2. Data file pruning: Within the target bucket, identifies only the data files that contain the primary key value.

3. Block-level range filtering: Applies accurate filtering based on primary key value distribution within file blocks, extracting only the blocks that contain the target value.

Optimized SQL query and analysis plans

Each bucket in a Delta table stores data that is unique and sorted by primary key value. The SQL Optimizer exploits these properties to eliminate expensive operations:

After eliminating DISTINCT, sorting, and global shuffling, query performance improves by 100%.