Hudi Data Table Compaction Specifications

Updated on 2024-09-10 GMT+08:00

View PDF

Updated data in the mor table is written in the form of row-store logs. When the logs are read, they need to be combined based on the primary key and are row-stored. As a result, the log reading efficiency is much lower than that of the parquet. Hudi uses compaction to compress logs into parquet files to improve the log read performance.

rules

If data is continuously written to a table, perform the compaction at least once within 24 hours.
For the MOR table, the compaction operation must be performed at least once a day, regardless of whether the MOR table is written in streaming mode or in batches. If compaction is not performed for a long time, the size of logs in the Hudi table will increase. In this case, the following problems will occur:
- Hudi table reading is slow and requires a lot of resources. This is because reading MOR tables involves log merging, which consumes a lot of resources and is slow.
- A long-term compaction process consumes a lot of resources and is prone to OOM.
- If no Compaction operation is performed to generate Parquet files of the new version, the files of the old version cannot be cleared by the Clean operation, increasing the storage pressure.
The ratio of CPU to memory is 1:4 to 1:8.
A Compaction job merges the data in the existing parquet file and the data in the new log file, which consumes a lot of memory resources. Based on the previous table design specifications and the actual traffic fluctuation, it is recommended that the ratio of CPU to memory for a Compaction job be 1. 4 to 1:8, ensuring stable running of Compaction jobs. If the OOM problem occurs in the Compaction, increase the memory usage.

The suggestion

The Compaction performance is improved by increasing the number of concurrent requests.
If the ratio of CPU to memory is proper, compaction jobs are stable and a single compaction task can run stably. However, the overall running duration of the Compaction depends on the number of files processed by the Compaction and the number of allocated CPU cores (concurrency capability). Therefore, the Compaction performance can be improved by increasing the number of CPU cores for the Compaciton job. (Note that the ratio of CPU to memory must be ensured when the CPU is increased.)
Hudi tables use asynchronous compaction.
To ensure the stable running of the streaming data import job, ensure that the streaming data import job does not perform other tasks during the real-time data import process. For example, Flink performs Compaction when writing data to Hudi. This seems to be a good solution, that is, completing the storage and compaction. However, the compaction operation consumes a lot of memory and I/O, and has the following impact on the streaming database import job:
- Increased end-to-end latency: Compaction increases the write latency because Compaction is more time-consuming than importing data to the database.
- Job instability: Compaction brings more instability to the stock-in job. Compaction OOM will cause the entire job to fail.
You are advised to perform the compaction every 2 to 4 hours.
Compaction is an important and mandatory maintenance method for MOR tables. For real-time tasks, the compaction process must be decoupled from the real-time tasks. Spark tasks are scheduled periodically to complete asynchronous compaction. The key point of this solution is how to set the period properly. If the period is too short, Spark tasks may be idle. If the period is too long, too many Compaction Plans may be accumulated and not executed. As a result, Spark tasks take a long time and downstream read job latency is long. In this scenario, the following suggestions are provided: Based on the cluster resource usage, asynchronous Compaction jobs can be scheduled and executed every two or four hours. This is a basic solution for maintaining MOR tables.

Spark is used to perform compaction asynchronously. Flink is not used to perform compaction.

The recommended solution for Flink to write hudi is that Flink only writes data and generates compaction plans. A separate Spark job asynchronously executes compaction, clean, and archive. Compaction plan generation is lightweight and has negligible impact on Flink write jobs.

The procedure for implementing the preceding solution is as follows:

Flink only writes data and generates compaction plans.

Add the following parameters to the table creation statement of the Flink stream task to ensure that only the compaction plan is generated when the Flink task writes data to the Hudi.

'compaction.async.enabled'='false' //Disable the Flink from executing the Compaction task.
'compaction.schedule.enabled' ='true' //Enable compaction plan generation.
'compaction.delta_commits' = '5' //By default, the MOR table attempts to generate a compaction plan after five checkpoints. This parameter needs to be adjusted based on the site requirements.
'clean.async.enabled' = 'false' //Disable the Clean operation.
'hoodie.archive.automatic' = 'false' //Disable the Archive operation.

Spark executes the Compaction plan offline, and performs the Clean and Archive operations.

On the scheduling platform (Huawei DataArts can be used), run a scheduled offline task to enable Spark to execute the Compaction plan and clean and archive the Hudi table.

set hoodie.archive.automatic = false;
set hoodie.clean.automatic = false;
set hoodie.compact.inline = true;
set hoodie.run.compact.only.inline=true;
The set hoodie.cleaner.commits.retained = 500; // clean retains the data files corresponding to the latest 500 deltacommits on the timeline. The files corresponding to the deltacommits in the earlier version will be deleted. The value must be greater than the value of compaction.delta_commits. Adjust the value based on the site requirements.
set hoodie.keep.max.commits = 700; // timeline retains a maximum of 700 deltacommits.
The set hoodie.keep.min.commits = 501; // timeline retains at least 500 deltacommits. The value must be greater than the value of hoodie.cleaner.commits.retained. Adjust the value based on the site requirements.
run compaction on <database name>. <table name>; // Executes the Compaction plan.
run clean on <database name>. <table name>; // Performing the Clean Operation
The run archivelog on <database name>.<table name>; // performs the Archive operation.

Asynchronous Compaction can serialize multiple tables into a job. Tables with similar resource configuration are put into a group. The resource configuration of the group of jobs is the resource required by the table that consumes the most resources.
For the people who are in the ·Asynchronous compaction is used for the HDI table. And to the · Use Spark to asynchronously execute Compaction, not... For the asynchronous Compaction task mentioned in, the following suggestions are provided:
- You do not need to develop an asynchronous compaction task for each Hudi table, which leads to high job development costs, explosion of cluster jobs, and ineffective use and release of cluster resources.
- Asynchronous compaction tasks can be completed by executing SparkSQL. Compaction, Clean, and Archive tasks of multiple Hudi tables can be executed in the same task. For example, asynchronous maintenance operations can be performed on table1 and table2.
```
set hoodie.clean.async = true;
set hoodie.clean.automatic = false;
set hoodie.compact.inline = true;
set hoodie.run.compact.only.inline=true;
set hoodie.cleaner.commits.retained = 500;
set hoodie.keep.min.commits = 501;
set hoodie.keep.max.commits = 700;
run compaction on <database name>. <table1>;
run clean on <database name>. <table1>;
run archivelog on <database name>.<table1>;
run compaction on <database name>.<table2>;
run clean on <database name>.<table2>;
run archivelog on <database name>.<table2>;
```

Parent topic: Hudi Data Table Management Operation Specifications

Previous topic: Hudi Data Table Management Operation Specifications

Next topic: Hudi Data Table Clean Specifications

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot