MaxCompute:Data organization optimization for PK Delta tables

Challenges

Delta tables support minute-level, near-real-time incremental data import. In high-traffic scenarios, this can cause a rapid increase in the number of small incremental files. This situation leads to the following problems:

• High storage costs and increased I/O load.

• Many small files also triggers frequent metadata updates.

• Slower analysis execution and inefficient data read and write operations.

For these reasons, a well-designed small file merging service, known as the Clustering service, is required to automatically optimize these scenarios.

Solution

The Clustering service is executed by the internal Storage Service of MaxCompute and is designed to merge small files. Clustering does not change any historical intermediate states of the data. This means that the service does not eliminate the intermediate historical state of any record.

Workflow

The overall workflow of the Clustering service is shown in the following figure.

1. Hierarchical merging

   The service uses clustering policies that are based on typical read/write business scenarios. The service periodically evaluates data files based on multiple dimensions, such as size and quantity, and performs hierarchical merges.

   1. Level 0 → Level 1: The smallest DeltaFiles from the initial write operation (blue data files in the figure) are merged into medium-sized DeltaFiles (yellow data files in the figure).

   2. Level 1 → Level 2: When the medium-sized DeltaFiles reach a certain scale, a higher-level merge is triggered to generate larger, optimized files (orange data files in the figure).

2. Avoiding read/write amplification

   1. Large file isolation: Data files that exceed a certain size, such as the T8 file in Bucket3, are excluded from the merge.

   2. Time span limit: Files with a large time span are not merged. This prevents the need to read large amounts of historical data that fall outside a query's time range during Time Travel or incremental queries.

   3. Automatic execution trigger: Each clustering operation reads and writes the data at least once. This consumes computing and I/O resources and causes some read/write amplification. To ensure that the Clustering service runs efficiently, the MaxCompute engine automatically triggers the operation based on the system status.

3. Concurrency and transactionality

   1. Concurrent execution: Because data is partitioned and stored by BucketIndex, the Clustering service runs concurrently at the bucket level. This significantly reduces the overall runtime.

   2. Transactional guarantee: The Clustering service interacts with the Meta Service to retrieve the list of tables or partitions to process. After the operation is complete, it passes information about the new and old data files to the Meta Service. The Meta Service plays a key role in this process. It is responsible for detecting transaction conflicts, coordinating the seamless update of metadata for new and old files, and securely reclaiming old data files.

Challenges

Delta tables support UPDATE and DELETE operations. These operations write new records to mark the previous state of old records, rather than modifying the old records in place. Many such operations can lead to:

• Data redundancy: Redundant records in intermediate states increase storage and computing costs.

• Lower query efficiency

Therefore, a well-designed Compaction service is required to eliminate intermediate states and optimize these scenarios.

Solution

Compaction merges selected data files, including BaseFiles and DeltaFiles. It combines multiple records with the same primary key and eliminates the intermediate UPDATE and DELETE states. Only the record with the latest state is retained. Finally, a new BaseFile is generated that contains only INSERT data.

Workflow

The overall workflow of the Compaction service is as follows.