gudladona commented on code in PR #20533:
URL: https://github.com/apache/hudi/pull/20533#discussion_r4494428615
##########
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/client/clustering/run/strategy/MultipleSparkJobExecutionStrategy.java:
##########
@@ -97,10 +102,58 @@
public abstract class MultipleSparkJobExecutionStrategy<T>
extends ClusteringExecutionStrategy<T, HoodieData<HoodieRecord<T>>,
HoodieData<HoodieKey>, HoodieData<WriteStatus>> {
+ /*
+ * Clustering builds one RDD per group and unions them into a single write
stage, so Spark ships the lineage of
+ * every group to every task of that stage. Java serialization writes an
instance once per stream however many
+ * closures reference it, but writes each distinct instance in full: an
object built per group is deserialized
+ * once per group by every task, and held by every running task. Whatever
does not depend on the group (table
+ * schema, reader context factory, reader properties, write config, writer
schema) is therefore built once per run
+ * and shared by all groups. The state below is driver-side only.
+ */
+ private transient HoodieWriteConfig groupWriteConfigSource;
+ private transient Map<Option<Long>, HoodieWriteConfig> groupWriteConfigs;
+ private transient Map<Pair<StructType, String>, StructType>
groupWriterSchemas;
Review Comment:
Thanks, done in cb47941c. The values the groups share now live in one
`ClusteringRunInvariants` holder. `performClustering` builds it before
submitting any group and clears it in `finally`.
- It's built after a subclass has settled the write config.
`SparkBinaryCopyClusteringExecutionStrategy` replaces the config before calling
`super.performClustering`, so the identity check is gone.
- `runClusteringForGroupAsync` and `runClusteringForGroupAsyncAsRow` are
back to their original signatures and read from the holder.
- The writer-schema cache stays keyed by dataset schema and table schema.
Both caches are concurrent maps, since groups run on the clustering thread
pool. They only memoize, so the holder stays fixed for the run. That's why it's
named "invariants" rather than "state".
- Called outside a run, `getGroupWriteConfig` and `getGroupWriterSchema`
build fresh values instead of failing.
No behavior change: every group still gets the same shared instances, and
`TestClusteringTaskBinaryPayloads` still passes.
--
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]
