Hudi Data Table Compaction Specifications

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.

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.

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.)

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.

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.

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.

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>;