gudladona opened a new pull request, #20533:
URL: https://github.com/apache/hudi/pull/20533
### Describe the issue this Pull Request addresses
Closes #20489
Clustering builds one RDD per clustering group and unions them into a single
write stage. Spark serializes the whole stage lineage into one task binary that
every task deserializes. Several objects that do not depend on the group were
built once per group, so each one was shipped to, deserialized by and held by
every task, once per group. The task binary and executor memory grew with
groups × concurrently running tasks.
### Summary and Changelog
Objects that do not depend on the group are built once per clustering run
and shared by every group. Java serialization writes a shared instance once per
stream, so each one now costs one copy per task, whatever the plan size. This
covers items (1)–(5) of #20489. Item (6), writing the group's
`RDD<InternalRow>` without the DataFrame round-trip, will follow in a separate
PR, as suggested in the review.
- `MultipleSparkJobExecutionStrategy`
- Resolves the table schema with meta fields once in `performClustering`,
instead of once per group. That was also one commit-metadata read on the driver
per group.
- Builds the `ReaderContextFactory` and the reader properties once. Each
`SparkReaderContextFactory` creates three broadcasts, so this also removes
three broadcasts per group. The non-row-writer path shares its factory and
reader schema the same way.
- Adds `getGroupWriteConfig(Option<Long> parquetMaxFileSize)`: one write
config per run, re-derived if the strategy's write config is replaced
(`SparkBinaryCopyClusteringExecutionStrategy` replaces it before clustering).
- Adds `getGroupWriterSchema(...)`: one bulk-insert writer schema per run.
- `SparkSortAndSizeExecutionStrategy`,
`SparkSingleFileSortExecutionStrategy`,
`SparkConsistentBucketClusteringExecutionStrategy`
- Use the shared write config and writer schema.
- `withBulkInsertParallelism(numOutputGroups)` is dropped, not moved after
`withProps`. It was overwritten by `withProps`, and nothing on the clustering
write paths reads `BULKINSERT_PARALLELISM_VALUE`: both paths pass
`numOutputGroups` explicitly. Making it effective would only make configs
differ per group.
- `HoodieDatasetBulkInsertHelper`
- New `writerSchema(datasetSchema, writeConfig)` and a `bulkInsert`
overload that takes the writer schema. The existing `bulkInsert` delegates to
it unchanged.
- `SQLConfInjectingRDD`
- RDDs that capture equal SQL configs share one configs map, keyed by
content, so a changed conf never gets a stale map.
- Tests
- `TestClusteringTaskBinaryPayloads` (new) runs the real
`SparkSortAndSizeExecutionStrategy` on a 2-group and an 8-group subset of one
plan. It serializes each write-stage RDD the way the closure serializer does,
and asserts that group-independent payloads do not grow with the number of
groups. It fails on master and passes with this change.
- `TestSQLConfInjectingRDD` (new) checks that the configs map is shared,
deduplicated in one task stream, never stale after a conf change, and still
applied to tasks.
`TestClusteringTaskBinaryPayloads`, before → after (2 groups → 8 groups):
| Payload | master | this PR |
|---|---|---|
| `HoodieSchema` | 3 → 9 | 2 → 2 |
| `HoodieWriteConfig` | 3 → 9 | 2 → 2 |
| `TypedProperties` | 45 → 129 | 31 → 31 |
| `ReaderContextFactory` | 2 → 8 | 1 → 1 |
| `StructType` | 6 → 24 | 3 → 3 |
| `SQLConfInjectingRDD` configs maps | 2 → 8 | 1 → 1 |
| Task binary growth per extra group | +73,013 B | +10,523 B |
The remaining growth per group is the group's own lineage: its RDDs,
partitions and file slices, plus the per-group `RDDScanExec` that (6) addresses.
### Impact
Performance: the write-stage task binary, task deserialization CPU and
executor heap no longer grow with the number of clustering groups for these
payloads. On a production table (Hudi 1.1.1, 4,000-group plans, 4 write tasks
per executor), the same changes in our fork took the task binary from 287.8 MiB
to 7.6 MiB and the executor live heap from 6.9–7.5 GB to 0.6–1.2 GB. The GC
share of task time fell from 18.9% to 1.4%. That fork also includes (6).
API: `runClusteringForGroupAsync` and `runClusteringForGroupAsyncAsRow`
(protected, in `MultipleSparkJobExecutionStrategy`) take the shared state as
extra parameters. Custom execution strategies that call or override them need
to pass it through. No in-repo subclass overrides them. Public APIs and configs
are unchanged.
### Risk Level
Low. The change shares immutable or read-only instances that were previously
copied per group:
- `HoodieSchema` and `StructType` are immutable.
- The write config is not mutated after it is built; nothing on the
bulk-insert paths calls `setValue` on it.
- Each task still deserializes its own copy, so nothing is shared across
tasks at runtime.
Verified locally:
- Spark 3.5 / Java 11: `TestClusteringTaskBinaryPayloads`,
`TestSQLConfInjectingRDD`, `TestSparkSortAndSizeClustering`,
`TestSparkConsistentBucketClustering`,
`TestHoodieSparkMergeOnReadTableClustering` and
`TestSparkBinaryCopyClusteringAndValidationMeta` pass. On my machine, a network
proxy breaks calls to the embedded timeline server, so the suites that use it
were run with it disabled and direct markers. That is the same setting
`TestClusteringTaskBinaryPayloads` uses.
- Spark 4.2 / Java 17 / Scala 2.13: builds, and
`TestClusteringTaskBinaryPayloads` and `TestSQLConfInjectingRDD` pass.
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
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]
