gudladona opened a new issue, #20489: URL: https://github.com/apache/hudi/issues/20489
### Bug Description ### Bug Description **What happened:** `MultipleSparkJobExecutionStrategy.performClustering` builds one RDD per clustering group and unions them into a single write stage. Spark serializes the whole stage lineage into one task binary, and every task in the stage deserializes it. So anything a group's lineage captures is shipped to every task, once for every group. Several objects that do not depend on the group are built and captured once per group. As a result: - the task binary grows linearly with the number of groups; - executor live memory grows with groups × concurrently running tasks; - task deserialization CPU for the whole stage grows with groups². The per-group payloads on master (`c3cbf4ea`): | # | Payload | Where | |---|---|---| | 1 | Table schema: resolved per group (re-reads commit metadata on the driver) and captured by the read closure. `HoodieSchema.readObject` re-parses the JSON on every deserialization. | `MultipleSparkJobExecutionStrategy.java:269-271`, `:326`; `HoodieSchema.java:3006-3014` | | 2 | `ReaderContextFactory`, built per group | `MultipleSparkJobExecutionStrategy.java:292`, `:323` | | 3 | `HoodieWriteConfig`, built per group with `newBuilder().withBulkInsertParallelism(n).withProps(getWriteConfig().getProps())`. Each one carries 13 `TypedProperties` copies (the config plus 12 sub-configs). | `SparkSortAndSizeExecutionStrategy.java:66-68`, `:90-92`; `SparkSingleFileSortExecutionStrategy.java:72-74`, `:100-102` | | 4 | `withBulkInsertParallelism(n)` is a no-op there: the `withProps(base)` that follows overwrites it with the base value | same lines as 3 | | 5 | Writer `StructType`, recomputed per call and captured by the `mapPartitions` closure | `HoodieDatasetBulkInsertHelper.scala:164` | | 6 | `RDDScanExec` plan per group from `createDataFrameFromRDD` | `MultipleSparkJobExecutionStrategy.java:342` | | 7 | Reader `TypedProperties`, built per group | `MultipleSparkJobExecutionStrategy.java:317` | | 8 | `SQLConfInjectingRDD.sqlConfigs = conf.getAllConfs`: a non-transient map copied into every RDD, and one RDD is created per group | `SQLConfInjectingRDD.scala:38`; `HoodieDatasetBulkInsertHelper.scala:98`, `:166` | Java serialization writes an instance once per stream, however many closures reference it, and a task's lineage is one stream. So sharing one instance across groups reduces each of these to one copy per task, whatever the plan size. No broadcast is needed. Related: #16441 (HUDI-7586) proposes using one schema for all clustering groups as a design cleanup. This issue adds the memory/CPU impact, a reproducing test, and the other payloads (2–8). **What you expected:** The write-stage task binary, and the per-task deserialized state, should not grow with the number of clustering groups. Only the per-group RDD lineage itself should. **Steps to reproduce:** 1. Check out the test-only branch [`gudladona/hudi:perf/clustering-per-group-task-payloads`](https://github.com/gudladona/hudi/tree/perf/clustering-per-group-task-payloads) (commit [`af9fd7b8`](https://github.com/gudladona/hudi/commit/af9fd7b81611), based on master `c3cbf4ea`). It adds [`TestClusteringTaskBinaryPayloads`](https://github.com/gudladona/hudi/blob/perf/clustering-per-group-task-payloads/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/client/clustering/run/strategy/TestClusteringTaskBinaryPayloads.java). 2. `mvn test -Dspark3.5 -pl hudi-spark-datasource/hudi-spark -Dtest=TestClusteringTaskBinaryPayloads` 3. The test does the following: - writes a 9-file-group COW table; - schedules one plan with `SparkSingleFileSortPlanStrategy` (one group per file group); - runs the real `SparkSortAndSizeExecutionStrategy.performClustering` on the first 2 and the first 8 groups; - serializes each write-stage RDD in one `ObjectOutputStream`, as the closure serializer does; - counts the distinct instances of each payload. On master it fails: ``` groups 2 8 HoodieSchema 3 9 +1.0 per group HoodieWriteConfig 3 9 +1.0 per group TypedProperties 45 129 +14.0 per group ReaderContextFactory 2 8 +1.0 per group StructType 6 24 +3.0 per group SQLConfInjectingRDD.sqlConfigs maps 2 8 +1.0 per group task binary bytes 245329 683407 +73013 bytes per group ``` ### Environment **Hudi version:** master `c3cbf4ea` (2026-10-31). We first observed this on 1.1.1 in production. **Query engine:** Spark 3.5 **Relevant configs:** - Spark clustering with the row writer (default). - `SparkSortAndSizeExecutionStrategy`. - Plans of 1,500–4,000 groups. - Executors with 4 cores, so 4 concurrent write tasks. ### Logs and Stack Trace **Production impact (1.1.1, a 4,000-group plan, executor with 4 write tasks running).** Measured with heap-dump retained sets, class histograms after a real Full GC, and JFR: | Per-group object | Instances on one executor | Retained | |---|---|---| | Table schema | 16,000 (= 4,000 groups × 4 tasks) | ~2.6 GB | | `HoodieWriteConfig` | 16,004 | ~2.9 GB (99.7% of all `ConcurrentHashMap$Node`) | | Writer `StructType` (via the `bulkInsert` closures) | 16,000 closures | ~0.36 GB | | `RDDScanExec` plans | 16,000 | ~0.41 GB | | `SQLConfInjectingRDD.sqlConfigs` maps | 16,000 | ~0.1 GB | - Together these were ~6.3 GB of a ~7.3 GB live heap during the write stage. - GC took 18.9% of task time. - Task deserialization was 16% of task CPU. - The driver logged `Broadcasting large task binary with size 287.8 MiB`. - One executor was removed after missing heartbeats during a long GC. **After sharing one instance of each across groups** (same table and plan size, in our fork): | | Before | After | |---|---|---| | Task binary | 287.8 MiB | 7.6 MiB | | Live heap per executor | 6.9–7.5 GB | 0.6–1.2 GB | | GC share of task time | 18.9% | 1.4% | | Task deserialization | 3.16 s CPU per task | 0.17 s | | Stage wall time | ~21 min | ~13 min | These savings let us cut the executor heap from 10 GB to 4 GB. **Proposed fix.** We are happy to send a PR; the test above turns green with it. 1. Resolve the table schema once in `performClustering` and pass the same `HoodieSchema` to every group (1). 2. Build `ReaderContextFactory` and the reader properties once (2, 7). 3. Use one shared write config per distinct (bulk-insert parallelism, max file size), built exactly as today (3). Fix or drop the ineffective `withBulkInsertParallelism` (4). 4. Add a `HoodieDatasetBulkInsertHelper.bulkInsert` overload that takes the writer schema, so clustering passes one instance (5). 5. `SQLConfInjectingRDD`: share the configs map across RDDs built from the same `SQLConf` contents (8). 6. Optional, larger: write the group's `RDD<InternalRow>` without the DataFrame round-trip (6). ### Environment **Hudi version: 1.1.x** **Ingestion engine: Spark ** ### Logs and Stack Trace _No response_ -- 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]
