voonhous commented on code in PR #20381: URL: https://github.com/apache/hudi/pull/20381#discussion_r4311189416
########## website/releases/release-1.1.md: ########## @@ -2,17 +2,57 @@ title: "Release 1.1" layout: releases toc: true -last_modified_at: 2024-05-02T18:00:00-08:00 +last_modified_at: 2026-10-17T18:59:30 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; This page contains release notes for all Apache Hudi 1.1.x releases, including: +- [Release 1.1.2](#release-112) - [Release 1.1.1](#release-111) --- +## [Release 1.1.2](https://github.com/apache/hudi/tree/release-1.1.2) {#release-112} + +## Migration Guide + +* This release (1.1.2) keeps the same table version (`9`) as [1.1.1](#release-111), so no table upgrade is needed if you are on 1.1.1. +* If migrating from an older release, please check the [1.1.1 release notes](#release-111) and the upgrade instructions from each older release in sequence. + +## Bundle Updates + +The same engine versions as 1.1.1 are supported: Spark 3.3–4.0 and Flink 1.17–1.20 and 2.0. See the [downloads](/releases/download#release-112) for the 1.1.2 bundles. + +## Behavior Changes + +* Flink upserts with CDC enabled now use blocking instant generation to preserve instant ordering, including on Merge-on-Read tables ([#18206](https://github.com/apache/hudi/pull/18206)). +* Compaction and log-compaction planning now limit concurrent partition scans with `hoodie.compaction.plan.parallelism`. The default is `200`, capped by the number of selected partitions; the value must be positive ([#20024](https://github.com/apache/hudi/pull/20024)). +* Timeline compaction adds `hoodie.timeline.compaction.target.file.max.bytes`, defaulting to `1048576000` bytes (1000 MiB), to limit each archived timeline file ([#17784](https://github.com/apache/hudi/pull/17784)). +* The new common writer option `hoodie.write.ignore.failed` defaults to `true` for backward compatibility. Flink's existing `write.ignore.failed` still defaults to `false` and now accepts the common option as a fallback key. With failure ignoring disabled, write-status errors fail the write earlier. Enabling it can allow checkpoints to proceed despite write-status errors. Review Comment: **major:** #12150 (`8734ffbe8eb6`) also removed the Flink coordinator's rollback: in 1.1.2 `StreamWriteOperatorCoordinator.doCommit` always commits and only logs write-status errors, whatever `write.ignore.failed` says. The flag now only decides whether write handles throw on record write exceptions, so errors that are only recorded (e.g. the partition-path mismatch in `HoodieAppendHandle.writeToBuffer`) are committed even with `false`. Could we add a line that Flink no longer rolls back an instant whose write statuses contain errors? ########## website/versioned_docs/version-1.1.2/flink_tuning.md: ########## @@ -0,0 +1,125 @@ +--- +title: Flink Tuning Guide +toc: true +--- + +## Global Configurations +When using Flink, you can set some global configurations in `$FLINK_HOME/conf/flink-conf.yaml` + +### Parallelism + +| Option Name | Default | Type | Description | +| ----------- | ------- | ------- | ------- | +| `taskmanager.numberOfTaskSlots` | `1` | `Integer` | The number of parallel operator or user function instances that a single TaskManager can run. We recommend setting this value > 4, and the actual value needs to be set according to the amount of data | +| `parallelism.default` | `1` | `Integer` | The default parallelism used when no parallelism is specified anywhere (default: 1). For example, If the value of [`write.bucket_assign.tasks`](#parallelism-1) is not set, this value will be used | + +### Memory + +| Option Name | Default | Type | Description | +| ----------- | ------- | ------- | ------- | +| `jobmanager.memory.process.size` | `(none)` | `MemorySize` | Total Process Memory size for the JobManager. This includes all the memory that a JobManager JVM process consumes, consisting of Total Flink Memory, JVM Metaspace, and JVM Overhead | +| `taskmanager.memory.task.heap.size` | `(none)` | `MemorySize` | Task Heap Memory size for TaskExecutors. This is the size of JVM heap memory reserved for write cache | +| `taskmanager.memory.managed.size` | `(none)` | `MemorySize` | Managed Memory size for TaskExecutors. This is the size of off-heap memory managed by the memory manager, reserved for sorting and RocksDB state backend. If you choose RocksDB as the state backend, you need to set this memory | + +### Checkpoint + +| Option Name | Default | Type | Description | +| ----------- | ------- | ------- | ------- | +| `execution.checkpointing.interval` | `(none)` | `Duration` | Setting this value as `execution.checkpointing.interval = 150000ms`, 150000ms = 2.5min. Configuring this parameter is equivalent to enabling the checkpoint | +| `state.backend` | `(none)` | `String` | The state backend to be used to store state. We recommend setting store state as `rocksdb` : `state.backend: rocksdb` | +| `state.backend.rocksdb.localdir` | `(none)` | `String` | The local directory (on the TaskManager) where RocksDB puts its files | +| `state.checkpoints.dir` | `(none)` | `String` | The default directory used for storing the data files and meta data of checkpoints in a Flink supported filesystem. The storage path must be accessible from all participating processes/nodes(i.e. all TaskManagers and JobManagers), like hdfs and oss path | +| `state.backend.incremental` | `false` | `Boolean` | Option whether the state backend should create incremental checkpoints, if possible. For an incremental checkpoint, only a diff from the previous checkpoint is stored, rather than the complete checkpoint state. If store state is setting as `rocksdb`, recommending to turn on | + +## Table Options + +Flink SQL jobs can be configured through options in the `WITH` clause. +The actual datasource level configs are listed below. + +### Memory + +:::note +When optimizing memory, we need to pay attention to the memory configuration +and the number of taskManagers, parallelism of write tasks (write.tasks : 4) first. After confirm each write task to be +allocated with enough memory, we can try to set these memory options. +::: + +| Option Name | Description | Default | Remarks | +| ----------- | ------- | ------- | ------- | +| `write.task.max.size` | Maximum memory in MB for a write task, when the threshold hits, it flushes the max size data bucket to avoid OOM. Default `1024MB` | `1024D` | The memory reserved for write buffer is `write.task.max.size` - `compaction.max_memory`. When total buffer of write tasks reach the threshold, the largest buffer in the memory will be flushed | +| `write.batch.size` | In order to improve the efficiency of writing, Flink write task will cache data in buffer according to the write bucket until the memory reaches the threshold. When reached threshold, the data buffer would be flushed out. Default `64MB` | `64D` | Recommend to use the default settings | +| `write.log_block.size` | The log writer of Hudi will not flush the data immediately after receiving data. The writer flush data to the disk in the unit of `LogBlock`. Before `LogBlock` reached threshold, records will be buffered in the writer in form of serialized bytes. Default `128MB` | `128` | Recommend to use the default settings | +| `write.merge.max_memory` | If write type is `COPY_ON_WRITE`, Hudi will merge the incremental data and base file data. The incremental data will be cached and spilled to disk. this threshold controls the max heap size that can be used. Default `100MB` | `100` | Recommend to use the default settings | +| `compaction.max_memory` | Same as `write.merge.max_memory`, but occurs during compaction. Default `100MB` | `100` | If it is online compaction, it can be turned up when resources are sufficient, such as setting as `1024MB` | + +### Parallelism + +| Option Name | Description | Default | Remarks | +| ----------- | ------- | ------- | ------- | +| `write.tasks` | The parallelism of writer tasks. Each write task writes 1 to `N` buckets in sequence. Default `4` | `4` | Increases the parallelism has no effect on the number of small files | +| `write.bucket_assign.tasks` | The parallelism of bucket assigner operators. No default value, using Flink `parallelism.default` | [`parallelism.default`](#parallelism) | Increases the parallelism also increases the number of buckets, thus the number of small files (small buckets) | +| `write.index_boostrap.tasks` | The parallelism of index bootstrap. Increasing parallelism can speed up the efficiency of the bootstrap stage. The bootstrap stage will block checkpointing. Therefore, it is necessary to set more checkpoint failure tolerance times. Default using Flink `parallelism.default` | [`parallelism.default`](#parallelism) | It only take effect when `index.bootsrap.enabled` is `true` | +| `read.tasks` | The parallelism of read operators (batch and stream). Default `4` | `4` | | +| `compaction.tasks` | The parallelism of online compaction. Default `4` | `4` | `Online compaction` will occupy the resources of the write task. It is recommended to use [`offline compaction`](compaction/#flink-offline-compaction) | + +### Compaction + +:::note +These are options only for `online compaction`. +::: + +:::note +Turn off online compaction by setting `compaction.async.enabled` = `false`, but we still recommend turning on `compaction.schedule.enable` for the writing job. You can then execute the compaction plan by [`offline compaction`](#compaction). +::: + +| Option Name | Description | Default | Remarks | +| ----------- | ------- | ------- | ------- | +| `compaction.schedule.enabled` | Whether to generate compaction plan periodically | `true` | Recommend to turn it on, even if `compaction.async.enabled` = `false` | +| `compaction.async.enabled` | Async Compaction, enabled by default for MOR | `true` | Turn off `online compaction` by turning off this option | +| `compaction.trigger.strategy` | Strategy to trigger compaction | `num_commits` | Options are `num_commits`: trigger compaction when reach N delta commits; `time_elapsed`: trigger compaction when time elapsed > N seconds since last compaction; `num_and_time`: trigger compaction when both `NUM_COMMITS` and `TIME_ELAPSED` are satisfied; `num_or_time`: trigger compaction when `NUM_COMMITS` or `TIME_ELAPSED` is satisfied. | +| `compaction.delta_commits` | Max delta commits needed to trigger compaction, default `5` commits | `5` | -- | +| `compaction.delta_seconds` | Max delta seconds time needed to trigger compaction, default `1` hour | `3600` | -- | +| `compaction.max_memory` | Max memory in MB for compaction spillable map, default `100MB` | `100` | If your have sufficient resources, recommend to adjust to `1024MB` | +| `compaction.target_io` | Target IO per compaction (both read and write), default `500GB`| `512000` | -- | + +## Memory Optimization + +### MOR + +1. [Setting Flink state backend to `rocksdb`](#checkpoint) (the default `in memory` state backend is very memory intensive). +2. If there is enough memory, `compaction.max_memory` can be set larger (`100MB` by default, and can be adjust to `1024MB`). +3. Pay attention to the memory allocated to each write task by taskManager to ensure that each write task can be allocated to the + desired memory size `write.task.max.size`. For example, taskManager has `4GB` of memory running two streamWriteFunction, so each write task + can be allocated with `2GB` memory. Please reserve some buffers because the network buffer and other types of tasks on taskManager (such as bucketAssignFunction) will also consume memory. +4. Pay attention to the memory changes of compaction. `compaction.max_memory` controls the maximum memory that each task can be used when compaction tasks read + logs. `compaction.tasks` controls the parallelism of compaction tasks. + +### COW + +1. [Setting Flink state backend to `rocksdb`](#checkpoint) (the default `in memory` state backend is very memory intensive). +2. Increase both `write.task.max.size` and `write.merge.max_memory` (`1024MB` and `100MB` by default, adjust to `2014MB` and `1024MB`). +3. Pay attention to the memory allocated to each write task by taskManager to ensure that each write task can be allocated to the + desired memory size `write.task.max.size`. For example, taskManager has `4GB` of memory running two write tasks, so each write task + can be allocated with `2GB` memory. Please reserve some buffers because the network buffer and other types of tasks on taskManager (such as `BucketAssignFunction`) will also consume memory. + + +## Write Rate Limit + +In the existing data synchronization, `snapshot data` and `incremental data` are send to kafka first, and then streaming write +to Hudi by Flink. Because the direct consumption of `snapshot data` will lead to problems such as high throughput and serious +disorder (writing partition randomly), which will lead to write performance degradation and throughput glitches. At this time, +the `write.rate.limit` option can be turned on to ensure smooth writing. + +### Options + +| Option Name | Required | Default | Remarks | +| ----------- | ------- | ------- | ------- | +| `write.rate.limit` | `false` | `0` | Turn off by default | + +## Streaming write reliability in 1.1.2 + +Version 1.1.2 fixes streaming write recovery after global failover and task retries, and reclaims memory from inactive write buckets when the buffer pool is exhausted. It also reduces repeated timeline loading and schema validation during writes. + +Keep `write.ignore.failed` at its default of `false` to fail on write-status errors. The common key `hoodie.write.ignore.failed` is accepted as a fallback; setting either option to `true` can allow checkpoints to complete despite write-status errors. Review Comment: **major:** Same root cause as the release-notes bullet: since #12150 the coordinator commits regardless of the flag, so `false` no longer fails on write-status errors in general; it makes record write exceptions fail the task eagerly. Could we reword along these lines? ```suggestion Keep `write.ignore.failed` at its default of `false` so that record write exceptions fail the task instead of being recorded in the write status. The common key `hoodie.write.ignore.failed` is accepted as a fallback. Since 1.1.2 the coordinator no longer rolls back an instant whose write statuses contain errors; such errors are logged and the instant is committed. ``` ########## website/versioned_docs/version-1.1.2/compaction.md: ########## @@ -0,0 +1,263 @@ +--- +title: Compaction +summary: "In this page, we describe async compaction in Hudi." +toc: true +toc_min_heading_level: 2 +toc_max_heading_level: 4 +last_modified_at: 2026-10-17T18:46:05 +--- +## Background + +Compaction is a table service employed by Hudi specifically in Merge-on-Read (MOR) tables to merge updates from row-based log +files to the corresponding columnar-based base file periodically to produce a new version of the base file. Compaction is +not applicable to Copy-on-Write (COW) tables and only applies to MOR tables. + +### Why MOR tables need compaction? + +To understand the significance of compaction in MOR tables, it is helpful to understand the MOR table layout first. In Hudi, +data is organized in terms of [file groups](https://hudi.apache.org/docs/storage_layouts/). Each file group in a MOR table +consists of a base file and one or more log files. Typically, during writes, inserts are stored in the base file, and updates +are appended to log files. + + +_Figure: MOR table file layout showing different file groups with base data file and log files_ + +During the compaction process, updates from the log files are merged with the base file to form a new version of the +base file as shown below. Since MOR is designed to be write-optimized, on new writes, after index tagging is complete, +Hudi appends the records pertaining to each file groups as log blocks in log files. There is no synchronous merge +happening during write, resulting in a lower write amplification and better write latency. In contrast, on new writes to a +COW table, Hudi combines the new writes with the older base file to produce a new version of the base file resulting in +a higher write amplification and higher write latencies. + + +_Figure: Compaction on a given file group_ + +While serving the read query(snapshot read), for each file group, records in base file and all its corresponding log +files are merged together and served. And hence the read latency for MOR snapshot query might be higher compared to +COW table since there is no merge involved in case of COW at read time. Compaction takes care of merging the updates from +log files with the base file at regular intervals to bound the growth of log files and to ensure the read latencies do not +spike up. + +## Compaction Architecture + +There are two steps to compaction. + +- **Compaction Scheduling**: In this step, Hudi scans the partitions and selects file slices to be compacted. A compaction plan is finally written to Hudi timeline. +- **Compaction Execution**: In this step the compaction plan is read and file slices are compacted. + +### Strategies in Compaction Scheduling + +There are two strategies involved in scheduling the compaction: + +- Trigger Strategy: Determines how often to trigger scheduling of the compaction. +- Compaction Strategy: Determines which file groups to compact. + +Hudi provides various options for both these strategies as discussed below. + +#### Incremental Scheduling + +Hudi supports incremental scheduling for compaction operations, which significantly improves performance on tables with a large number of partitions. Instead of scanning all partitions during each compaction scheduling run, incremental scheduling only processes partitions that have changed since the last completed compaction. + +This feature is enabled by default via `hoodie.table.services.incremental.enabled`. When enabled, compaction scheduling will: + +1. Identify partitions that have been modified since the last completed compaction by analyzing commit metadata within the time window between the last completed compaction and the current scheduling instant +2. Include any partitions that were marked as missing from previous scheduling runs (e.g., due to IO limits) +3. Only scan and process those incremental partitions +4. Fall back to scanning all partitions if the last completed compaction instant cannot be found (e.g., due to archival) or if an exception occurs during incremental partition retrieval + +For tables with many partitions, this optimization can dramatically reduce scheduling overhead and improve overall job stability. + +#### Trigger Strategies + +| Config Name | Default | Description || +|----------------------------------------------------|-------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------| +| hoodie.compact.inline.trigger.strategy | NUM_COMMITS (Optional) | org.apache.hudi.table.action.compact.CompactionTriggerStrategy: Controls when compaction is scheduled.<br />`Config Param: INLINE_COMPACT_TRIGGER_STRATEGY` <br/> +<ul><li>`NUM_COMMITS`: triggers compaction when there are at least N delta commits after last completed compaction.</li><li>`NUM_COMMITS_AFTER_LAST_REQUEST`: triggers compaction when there are at least N delta commits after last completed or requested compaction.</li><li>`TIME_ELAPSED`: triggers compaction after N seconds since last compaction.</li><li>`NUM_AND_TIME`: triggers compaction when both there are at least N delta commits and N seconds elapsed (both must be satisfied) after last completed compaction.</li><li>`NUM_OR_TIME`: triggers compaction when both there are at least N delta commits or N seconds elapsed (either condition is satisfied) after last completed compaction.</li></ul> | + +#### Compaction Strategies + +| Config Name | Default | Description | +|----------------------------|---------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| hoodie.compaction.strategy | org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy (Optional) | Compaction strategy decides which file groups are picked up for compaction during each compaction run. By default. Hudi picks the log file with most accumulated unmerged data. <br /><br />`Config Param: COMPACTION_STRATEGY` | + +Available Strategies (Provide the full package name when using the strategy): <ul><li>`LogFileNumBasedCompactionStrategy`: +orders the compactions based on the total log files count, filters the file group with log files count greater than the +threshold and limits the compactions within a configured IO bound.</li><li>`LogFileSizeBasedCompactionStrategy`: orders +the compactions based on the total log files size, filters the file group which log files size is greater than the +threshold and limits the compactions within a configured IO bound.</li><li>`BoundedIOCompactionStrategy`: CompactionStrategy +which looks at total IO to be done for the compaction (read + write) and limits the list of compactions to be under a +configured limit on the IO.</li><li>`BoundedPartitionAwareCompactionStrategy`:This strategy ensures that the last N partitions +are picked up even if there are later partitions created for the table. lastNPartitions is defined as the N partitions before +the currentDate. currentDay = 2018/01/01 The table has partitions for 2018/02/02 and 2018/03/03 beyond the currentDay This +strategy will pick up the following partitions for compaction : (2018/01/01, allPartitionsInRange[(2018/01/01 - lastNPartitions) +to 2018/01/01), 2018/02/02, 2018/03/03)</li><li>`DayBasedCompactionStrategy`:This strategy orders compactions in reverse +order of creation of Hive Partitions. It helps to compact data in latest partitions first and then older capped at the +Total_IO allowed.</li><li>`UnBoundedCompactionStrategy`: UnBoundedCompactionStrategy will not change ordering or filter +any compaction. It is a pass-through and will compact all the base files which has a log file. This usually means +no-intelligence on compaction.</li><li>`UnBoundedPartitionAwareCompactionStrategy`:UnBoundedPartitionAwareCompactionStrategy is a custom UnBounded Strategy. This will filter all the partitions that +are eligible to be compacted by a \{@link BoundedPartitionAwareCompactionStrategy} and return the result. This is done +so that a long running UnBoundedPartitionAwareCompactionStrategy does not step over partitions in a shorter running +BoundedPartitionAwareCompactionStrategy. Essentially, this is an inverse of the partitions chosen in +BoundedPartitionAwareCompactionStrategy</li></ul> + +:::note +Please refer to [advanced configs](https://hudi.apache.org/docs/next/configurations#Compaction-Configs) for more details. +::: + +## Ways to trigger Compaction + +### Inline + +By default, compaction is run asynchronously. + +If latency of ingesting records is important for you, you are most likely using Merge-on-Read tables. +Merge-on-Read tables store data using a combination of columnar (e.g parquet) + row based (e.g avro) file formats. +Updates are logged to delta files & later compacted to produce new versions of columnar files. +To improve ingestion latency, Async Compaction is the default configuration. + +If immediate read performance of a new commit is important for you, or you want simplicity of not managing separate compaction jobs, +you may want synchronous inline compaction, which means that as a commit is written it is also compacted by the same job. + +For this deployment mode, please use `hoodie.compact.inline = true` for Spark Datasource and Spark SQL writers. For +Hudi Streamer sync once mode inline compaction can be achieved by passing the flag `--disable-compaction` (Meaning to +disable async compaction). Further in Hudi Streamer when both +ingestion and compaction is running in the same spark context, you can use resource allocation configuration +in Hudi Streamer CLI such as (`--delta-sync-scheduling-weight`, +`--compact-scheduling-weight`, `--delta-sync-scheduling-minshare`, and `--compact-scheduling-minshare`) +to control executor allocation between ingestion and compaction. + +### Async & Offline Compaction models + +There are a couple of ways here to trigger compaction . + +#### Async execution within the same process + +In streaming ingestion write models like Hudi Streamer +continuous mode, Flink and Spark Streaming, async compaction is enabled by default and runs alongside without blocking +regular ingestion. + +##### Spark Structured Streaming + +Compactions are scheduled and executed asynchronously inside the +streaming job.Here is an example snippet in java + +```properties +import org.apache.hudi.DataSourceWriteOptions; +import org.apache.hudi.HoodieDataSourceHelpers; +import org.apache.hudi.config.HoodieCompactionConfig; +import org.apache.hudi.config.HoodieWriteConfig; + +import org.apache.spark.sql.streaming.OutputMode; +import org.apache.spark.sql.streaming.ProcessingTime; + DataStreamWriter<Row> writer = streamingInput.writeStream().format("org.apache.hudi") + .option("hoodie.datasource.write.operation", operationType) + .option("hoodie.datasource.write.table.type", tableType) + .option("hoodie.datasource.write.recordkey.field", "_row_key") + .option("hoodie.datasource.write.partitionpath.field", "partition") + .option("hoodie.table.ordering.fields", "timestamp") + .option("hoodie.compact.inline.max.delta.commits", "10") + .option("hoodie.datasource.compaction.async.enable", "true") + .option("hoodie.table.name", tableName).option("checkpointLocation", checkpointLocation) + .outputMode(OutputMode.Append()); + writer.trigger(new ProcessingTime(30000)).start(tablePath); +``` + +##### Hudi Streamer Continuous Mode + +Hudi Streamer provides continuous ingestion mode where a single long running spark application +ingests data to Hudi table continuously from upstream sources. In this mode, Hudi supports managing asynchronous +compactions. Here is an example snippet for running in continuous mode with async compactions + +```properties +spark-submit --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.0.2,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.2 \ Review Comment: **minor:** This snapshot still pins `1.0.2` here and at line 212, plus `cleaning.md` (83, 107, 117, 126), `disaster_recovery.md` (151, 310) and the `--jars` examples in `metadata_indexing.md` and `clustering.md`. They come from the 1.1.1 copy; the 1.2.1 snapshot has none. Not blocking, but could we bump these to `1.1.2` like the ones already updated in `cli.md` and the quick starts? ```suggestion spark-submit --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.1.2,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.1.2 \ ``` ########## website/releases/release-1.1.md: ########## @@ -2,17 +2,57 @@ title: "Release 1.1" layout: releases toc: true -last_modified_at: 2024-05-02T18:00:00-08:00 +last_modified_at: 2026-10-17T18:59:30 --- import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; This page contains release notes for all Apache Hudi 1.1.x releases, including: +- [Release 1.1.2](#release-112) - [Release 1.1.1](#release-111) --- +## [Release 1.1.2](https://github.com/apache/hudi/tree/release-1.1.2) {#release-112} Review Comment: **nit:** Every other release heading links to `releases/tag/...`, and the 1.1.2 GitHub release was published on 2026-10-17. Feel free to ignore, but could we point it at the release page? ```suggestion ## [Release 1.1.2](https://github.com/apache/hudi/releases/tag/release-1.1.2) {#release-112} ``` ########## website/versioned_docs/version-1.1.2/compaction.md: ########## @@ -0,0 +1,263 @@ +--- +title: Compaction +summary: "In this page, we describe async compaction in Hudi." +toc: true +toc_min_heading_level: 2 +toc_max_heading_level: 4 +last_modified_at: 2026-10-17T18:46:05 +--- +## Background + +Compaction is a table service employed by Hudi specifically in Merge-on-Read (MOR) tables to merge updates from row-based log +files to the corresponding columnar-based base file periodically to produce a new version of the base file. Compaction is +not applicable to Copy-on-Write (COW) tables and only applies to MOR tables. + +### Why MOR tables need compaction? + +To understand the significance of compaction in MOR tables, it is helpful to understand the MOR table layout first. In Hudi, +data is organized in terms of [file groups](https://hudi.apache.org/docs/storage_layouts/). Each file group in a MOR table +consists of a base file and one or more log files. Typically, during writes, inserts are stored in the base file, and updates +are appended to log files. + + +_Figure: MOR table file layout showing different file groups with base data file and log files_ + +During the compaction process, updates from the log files are merged with the base file to form a new version of the +base file as shown below. Since MOR is designed to be write-optimized, on new writes, after index tagging is complete, +Hudi appends the records pertaining to each file groups as log blocks in log files. There is no synchronous merge +happening during write, resulting in a lower write amplification and better write latency. In contrast, on new writes to a +COW table, Hudi combines the new writes with the older base file to produce a new version of the base file resulting in +a higher write amplification and higher write latencies. + + +_Figure: Compaction on a given file group_ + +While serving the read query(snapshot read), for each file group, records in base file and all its corresponding log +files are merged together and served. And hence the read latency for MOR snapshot query might be higher compared to +COW table since there is no merge involved in case of COW at read time. Compaction takes care of merging the updates from +log files with the base file at regular intervals to bound the growth of log files and to ensure the read latencies do not +spike up. + +## Compaction Architecture + +There are two steps to compaction. + +- **Compaction Scheduling**: In this step, Hudi scans the partitions and selects file slices to be compacted. A compaction plan is finally written to Hudi timeline. +- **Compaction Execution**: In this step the compaction plan is read and file slices are compacted. + +### Strategies in Compaction Scheduling + +There are two strategies involved in scheduling the compaction: + +- Trigger Strategy: Determines how often to trigger scheduling of the compaction. +- Compaction Strategy: Determines which file groups to compact. + +Hudi provides various options for both these strategies as discussed below. + +#### Incremental Scheduling + +Hudi supports incremental scheduling for compaction operations, which significantly improves performance on tables with a large number of partitions. Instead of scanning all partitions during each compaction scheduling run, incremental scheduling only processes partitions that have changed since the last completed compaction. + +This feature is enabled by default via `hoodie.table.services.incremental.enabled`. When enabled, compaction scheduling will: + +1. Identify partitions that have been modified since the last completed compaction by analyzing commit metadata within the time window between the last completed compaction and the current scheduling instant +2. Include any partitions that were marked as missing from previous scheduling runs (e.g., due to IO limits) +3. Only scan and process those incremental partitions +4. Fall back to scanning all partitions if the last completed compaction instant cannot be found (e.g., due to archival) or if an exception occurs during incremental partition retrieval + +For tables with many partitions, this optimization can dramatically reduce scheduling overhead and improve overall job stability. + +#### Trigger Strategies + +| Config Name | Default | Description || +|----------------------------------------------------|-------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------| +| hoodie.compact.inline.trigger.strategy | NUM_COMMITS (Optional) | org.apache.hudi.table.action.compact.CompactionTriggerStrategy: Controls when compaction is scheduled.<br />`Config Param: INLINE_COMPACT_TRIGGER_STRATEGY` <br/> +<ul><li>`NUM_COMMITS`: triggers compaction when there are at least N delta commits after last completed compaction.</li><li>`NUM_COMMITS_AFTER_LAST_REQUEST`: triggers compaction when there are at least N delta commits after last completed or requested compaction.</li><li>`TIME_ELAPSED`: triggers compaction after N seconds since last compaction.</li><li>`NUM_AND_TIME`: triggers compaction when both there are at least N delta commits and N seconds elapsed (both must be satisfied) after last completed compaction.</li><li>`NUM_OR_TIME`: triggers compaction when both there are at least N delta commits or N seconds elapsed (either condition is satisfied) after last completed compaction.</li></ul> | + +#### Compaction Strategies + +| Config Name | Default | Description | +|----------------------------|---------------------------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| hoodie.compaction.strategy | org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy (Optional) | Compaction strategy decides which file groups are picked up for compaction during each compaction run. By default. Hudi picks the log file with most accumulated unmerged data. <br /><br />`Config Param: COMPACTION_STRATEGY` | + +Available Strategies (Provide the full package name when using the strategy): <ul><li>`LogFileNumBasedCompactionStrategy`: +orders the compactions based on the total log files count, filters the file group with log files count greater than the +threshold and limits the compactions within a configured IO bound.</li><li>`LogFileSizeBasedCompactionStrategy`: orders +the compactions based on the total log files size, filters the file group which log files size is greater than the +threshold and limits the compactions within a configured IO bound.</li><li>`BoundedIOCompactionStrategy`: CompactionStrategy +which looks at total IO to be done for the compaction (read + write) and limits the list of compactions to be under a +configured limit on the IO.</li><li>`BoundedPartitionAwareCompactionStrategy`:This strategy ensures that the last N partitions +are picked up even if there are later partitions created for the table. lastNPartitions is defined as the N partitions before +the currentDate. currentDay = 2018/01/01 The table has partitions for 2018/02/02 and 2018/03/03 beyond the currentDay This +strategy will pick up the following partitions for compaction : (2018/01/01, allPartitionsInRange[(2018/01/01 - lastNPartitions) +to 2018/01/01), 2018/02/02, 2018/03/03)</li><li>`DayBasedCompactionStrategy`:This strategy orders compactions in reverse +order of creation of Hive Partitions. It helps to compact data in latest partitions first and then older capped at the +Total_IO allowed.</li><li>`UnBoundedCompactionStrategy`: UnBoundedCompactionStrategy will not change ordering or filter +any compaction. It is a pass-through and will compact all the base files which has a log file. This usually means +no-intelligence on compaction.</li><li>`UnBoundedPartitionAwareCompactionStrategy`:UnBoundedPartitionAwareCompactionStrategy is a custom UnBounded Strategy. This will filter all the partitions that +are eligible to be compacted by a \{@link BoundedPartitionAwareCompactionStrategy} and return the result. This is done +so that a long running UnBoundedPartitionAwareCompactionStrategy does not step over partitions in a shorter running +BoundedPartitionAwareCompactionStrategy. Essentially, this is an inverse of the partitions chosen in +BoundedPartitionAwareCompactionStrategy</li></ul> + +:::note +Please refer to [advanced configs](https://hudi.apache.org/docs/next/configurations#Compaction-Configs) for more details. +::: + +## Ways to trigger Compaction + +### Inline + +By default, compaction is run asynchronously. + +If latency of ingesting records is important for you, you are most likely using Merge-on-Read tables. +Merge-on-Read tables store data using a combination of columnar (e.g parquet) + row based (e.g avro) file formats. +Updates are logged to delta files & later compacted to produce new versions of columnar files. +To improve ingestion latency, Async Compaction is the default configuration. + +If immediate read performance of a new commit is important for you, or you want simplicity of not managing separate compaction jobs, +you may want synchronous inline compaction, which means that as a commit is written it is also compacted by the same job. + +For this deployment mode, please use `hoodie.compact.inline = true` for Spark Datasource and Spark SQL writers. For +Hudi Streamer sync once mode inline compaction can be achieved by passing the flag `--disable-compaction` (Meaning to +disable async compaction). Further in Hudi Streamer when both +ingestion and compaction is running in the same spark context, you can use resource allocation configuration +in Hudi Streamer CLI such as (`--delta-sync-scheduling-weight`, +`--compact-scheduling-weight`, `--delta-sync-scheduling-minshare`, and `--compact-scheduling-minshare`) +to control executor allocation between ingestion and compaction. + +### Async & Offline Compaction models + +There are a couple of ways here to trigger compaction . + +#### Async execution within the same process + +In streaming ingestion write models like Hudi Streamer +continuous mode, Flink and Spark Streaming, async compaction is enabled by default and runs alongside without blocking +regular ingestion. + +##### Spark Structured Streaming + +Compactions are scheduled and executed asynchronously inside the +streaming job.Here is an example snippet in java + +```properties +import org.apache.hudi.DataSourceWriteOptions; +import org.apache.hudi.HoodieDataSourceHelpers; +import org.apache.hudi.config.HoodieCompactionConfig; +import org.apache.hudi.config.HoodieWriteConfig; + +import org.apache.spark.sql.streaming.OutputMode; +import org.apache.spark.sql.streaming.ProcessingTime; + DataStreamWriter<Row> writer = streamingInput.writeStream().format("org.apache.hudi") + .option("hoodie.datasource.write.operation", operationType) + .option("hoodie.datasource.write.table.type", tableType) + .option("hoodie.datasource.write.recordkey.field", "_row_key") + .option("hoodie.datasource.write.partitionpath.field", "partition") + .option("hoodie.table.ordering.fields", "timestamp") + .option("hoodie.compact.inline.max.delta.commits", "10") + .option("hoodie.datasource.compaction.async.enable", "true") + .option("hoodie.table.name", tableName).option("checkpointLocation", checkpointLocation) + .outputMode(OutputMode.Append()); + writer.trigger(new ProcessingTime(30000)).start(tablePath); +``` + +##### Hudi Streamer Continuous Mode + +Hudi Streamer provides continuous ingestion mode where a single long running spark application +ingests data to Hudi table continuously from upstream sources. In this mode, Hudi supports managing asynchronous +compactions. Here is an example snippet for running in continuous mode with async compactions + +```properties +spark-submit --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.0.2,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.2 \ +--class org.apache.hudi.utilities.streamer.HoodieStreamer \ +--table-type MERGE_ON_READ \ +--target-base-path <hudi_base_path> \ +--target-table <hudi_table> \ +--source-class org.apache.hudi.utilities.sources.JsonDFSSource \ +--source-ordering-field ts \ +--props /path/to/source.properties \ +--continous +``` + +#### Scheduling and Execution by a separate process + +For some use cases with long running table services, instead of having the regular writes block, users have the option to run +both steps of the compaction ([scheduling and execution](#compaction-architecture)) offline in a separate process altogether. +This allows for regular writers to not bother about these compaction steps and allows users to provide more resources for +the compaction job as needed. + +:::note +This model needs a lock provider configured for all jobs - the regular writer as well as the offline compaction job. +::: + +#### Scheduling inline and executing async + +In this model, it is possible for a Spark Datasource writer or a Flink job to just schedule the compaction inline ( that +will serialize the compaction plan in the timeline but will not execute it). And then a separate utility like +HudiCompactor or HoodieFlinkCompactor can take care of periodically executing the compaction plan. + +:::note +This model may need a lock provider **if** metadata table is enabled. +::: + +#### Hudi Compactor Utility + +Hudi provides a standalone tool to execute specific compactions asynchronously. Below is an example and you can read more in the [deployment guide](cli.md#compactions) +The compactor utility allows to do scheduling and execution of compaction. + +Example: + +```properties +spark-submit --packages org.apache.hudi:hudi-utilities-slim-bundle_2.12:1.0.2,org.apache.hudi:hudi-spark3.5-bundle_2.12:1.0.2 \ +--class org.apache.hudi.utilities.HoodieCompactor \ +--base-path <base_path> \ +--table-name <table_name> \ +--schema-file <schema_file> \ +--instant-time <compaction_instant> +``` + +Note, the `instant-time` parameter is now optional for the Hudi Compactor Utility. If using the utility without `--instant time`, +the spark-submit will execute the earliest scheduled compaction on the Hudi timeline. + +#### Hudi CLI + +Hudi CLI is yet another way to execute specific compactions asynchronously. Here is an example and you can read more in the [deployment guide](cli.md#compactions) + +Example: + +```properties +hudi:trips->compaction run --tableName <table_name> --parallelism <parallelism> --compactionInstant <InstantTime> +... +``` + +#### Flink Offline Compaction + +Offline compaction needs to submit the Flink task on the command line. The program entry is as follows: `hudi-flink-bundle_2.11-0.9.0-SNAPSHOT.jar` : +`org.apache.hudi.sink.compact.HoodieFlinkCompactor` + +```bash +# Command line +./bin/flink run -c org.apache.hudi.sink.compact.HoodieFlinkCompactor lib/hudi-flink-bundle_2.11-0.9.0.jar --path hdfs://xxx:9000/table +``` + +#### Options + +| Option Name | Default | Description | +|-------------------------------------|----------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| `--path` | `n/a **(Required)**` | The path where the target table is stored on Hudi | +| `--compaction-max-memory` | `100` (Optional) | The index map size of log data during compaction, 100 MB by default. If you have enough memory, you can turn up this parameter | +| `--schedule` | `false` (Optional) | whether to execute the operation of scheduling compaction plan. When the write process is still writing， turning on this parameter have a risk of losing data. Therefore, it must be ensured that there are no write tasks currently writing data to this table when this parameter is turned on | +| `--seq` | `LIFO` (Optional) | The order in which compaction tasks are executed. Executing from the latest compaction plan by default. `LIFO`: executing from the latest plan. `FIFO`: executing from the oldest plan. | +| `--service` | `false` (Optional) | Whether to start a monitoring service that checks and schedules new compaction task in configured interval. | +| `--min-compaction-interval-seconds` | `600(s)` (optional) | The checking interval for service mode, by default 10 minutes. | + +## Related Resources + +<h3>Blogs</h3> +[Apache Hudi Compaction](https://medium.com/@simpsons/apache-hudi-compaction-6e6383790234) +[Standalone HoodieCompactor Utility](https://medium.com/@simpsons/standalone-hoodiecompactor-utility-890198e4c539) + +### Compaction planning parallelism Review Comment: **nit:** This section lands after `## Related Resources` and its blog links, so the TOC nests it under Related Resources. Feel free to ignore, but could we move it up, e.g. just before `## Related Resources` or under `## Compaction Architecture`? ########## website/versioned_docs/version-1.1.2/configurations.md: ########## Review Comment: **nit:** The `hoodie.write.ignore.failed` row (line 962) sits above `hoodie.base.path`, out of the alphabetical order the generator emits. Since `IGNORE_FAILED` is not `markAdvanced()`, the generator would also list it in `basic_configurations.md`, where it is missing. Feel free to ignore, but could we move it to its alphabetical slot and add the same row to `basic_configurations.md`? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
