Hudi Clustering

Updated on 2025-02-22 GMT+08:00

View PDF

Introduction

Clustering reorganizes data layout to improve query performance without affecting the ingestion speed.

Architecture

Hudi provides different operations, such as insert, upsert, and bulk_insert, through its write client API to write data to a Hudi table. To weight between file size and speed of importing data into the data lake, Hudi provides hoodie.parquet.small.file.limit to configure the minimum file size. You can set it to 0 to force new data to be written to new file groups, or to a higher value to ensure that new data is "padded" to existing small file groups until it reaches the specified size, but this increases ingestion latency.

To support fast ingestion without affecting query performance, the clustering service is introduced to rewrite data to optimize the layout of Hudi data lake files.

The clustering service can run asynchronously or synchronously. It adds a new operation type called REPLACE, which will mark the clustering operation in the Hudi metadata timeline.

Clustering service is based on the MVCC design of Hudi to allow new data to be inserted. Clustering operations run in the background to reformat data layout, ensuring snapshot isolation between concurrent readers and writers.

Clustering is divided into two parts:

Scheduling clustering: Create a clustering plan using a pluggable clustering strategy.
1. Identify files that are eligible for clustering: Depending on the selected clustering strategy, the scheduling logic will identify the files eligible for clustering.
2. Group files that are eligible for clustering based on specific criteria. The data size of each group must be a multiple of targetFileSize. Grouping is a part of the strategy defined in the plan. Additionally, there is an option to control group size to improve parallelism and avoid shuffling large volumes of data.
3. Save the clustering plan to the timeline in Avro metadata format.
Execute clustering: Process the plan using an execution strategy to create new files and replace old files.
1. Read the clustering plan and obtain clusteringGroups that marks the file groups to be clustered.
2. Instantiate appropriate strategy class for each group using strategyParams (for example, sortColumns) and apply the strategy to rewrite data.
3. Create a REPLACE commit and update the metadata in HoodieReplaceCommitMetadata.

How to Execute Clustering

Spark SQL (Set the following parameters, trigger on data write)

hoodie.clustering.inline=true // The default value is false, meaning clustering is disabled by default.
hoodie.clustering.inline.max.commits=4 // The default value is 4, but you can adjust it based on the service scenario.
hoodie.clustering.plan.strategy.max.bytes.per.group=2147483648 // The default value is 2 GB, but you can adjust it based on the service scenario. Generally, no need to specify as the normal data amount under each file group does not exceed 2 GB.
hoodie.clustering.plan.strategy.max.num.groups=30 // The default value is 30, but you can adjust it based on the service scenario. Usually, adjust this parameter to adjust the data amount for each clustering plan (max.bytes.per.group x max.num.groups).
hoodie.clustering.plan.strategy.partition.regex.pattern=${Regular expression} // No default value. If not specified, clustering will reorganize data across all partitions.
hoodie.clustering.plan.strategy.small.file.limit=314572800 // The default value is 300 MB, but you can adjust it based on the service scenario. Files smaller than 300 MB in each partition will be selected for clustering.
hoodie.clustering.plan.strategy.sort.columns=${Column 1,...,Column N} // No default value, but you can set it based on the service scenario. Specify columns frequently used in queries and do not contain null.
hoodie.clustering.plan.strategy.target.file.max.bytes=1073741824 // The default value is 1 GB, but you can adjust it based on the service scenario. It is used to set the maximum size of files generated by clustering.

SparkDataSource (Set the following parameters in the option, trigger on data write)
option("hoodie.clustering.inline", "true").

option("hoodie.clustering.inline.max.commits", 4).

option("hoodie.clustering.plan.strategy.max.bytes.per.group", 2147483648).

option("hoodie.clustering.plan.strategy.max.num.groups", 30).

option("hoodie.clustering.plan.strategy.partition.regex.pattern", "${Regular expression}").

option("hoodie.clustering.plan.strategy.small.file.limit", 314572800).

option("hoodie.clustering.plan.strategy.sort.columns", "${Column 1,......,Column N}").

option("hoodie.clustering.plan.strategy.target.file.max.bytes", 1073741824).

Manually trigger clustering once.

Spark SQL (Set the following parameters, manually trigger once)

hoodie.clustering.inline=true
hoodie.clustering.inline.max.commits=4 // The default value is 4, but you can adjust it based on the service scenario.
hoodie.clustering.plan.strategy.max.bytes.per.group=2147483648 // The default value is 2 GB, but you can adjust it based on the service scenario. Generally, no need to specify as the normal data amount under each file group does not exceed 2 GB.
hoodie.clustering.plan.strategy.max.num.groups=30 // The default value is 30, but you can adjust it based on the service scenario. Usually, adjust this parameter to adjust the data amount for each clustering plan (max.bytes.per.group x max.num.groups).
hoodie.clustering.plan.strategy.partition.regex.pattern=${Regular expression} // No default value. If not specified, clustering will reorganize data across all partitions.
hoodie.clustering.plan.strategy.small.file.limit=314572800 // The default value is 300 MB, but you can adjust it based on the service scenario. Files smaller than 300 MB in each partition will be selected for clustering.
hoodie.clustering.plan.strategy.sort.columns=${Column 1,...,Column N} // No default value, but you can set it based on the service scenario. Specify columns frequently used in queries and do not contain null.
hoodie.clustering.plan.strategy.target.file.max.bytes=1073741824 // The default value is 1 GB, but you can adjust it based on the service scenario. It is used to set the maximum size of files generated by clustering.

Run the following SQL statement:

call run_clustering (table =>'${Table name}')

CAUTION:

Clustering sort columns cannot contain null values, due to Spark RDD limitations.
When the value of target.file.max.bytes is large, increase --executor-memory to avoid executor memory overflow.
Clean does not support cleaning residual files after clustering failures.
New files generated after clustering may vary in size, which could cause data skew.
Clustering does not support concurrency with upsert (write operation updates files waiting for clustering). If clustering is inflight, files under that FileGroup cannot be updated.
If there are unfinished clustering plans, subsequent write triggers generating compaction plans may fail. Execute clustering plans promptly.