Diveyam Mishra created SPARK-59174:
--------------------------------------
Summary: State-store instance metrics are shipped in every task
result
Key: SPARK-59174
URL: https://issues.apache.org/jira/browse/SPARK-59174
Project: Spark
Issue Type: Improvement
Components: SQL, Structured Streaming
Affects Versions: 4.0.4
Reporter: Diveyam Mishra
Stateful streaming operators (StateStoreWriter) currently create a separate
SQLMetric accumulator for every (state partition x instance metric x store
name) combination upfront on the physical plan. On executors, every task
deserializes and registers all of these accumulators with its TaskContext, and
returns all of them in the task result payload even though only one partition
was processed. For queries with high partition counts (e.g., 20,000 shuffle
partitions), this causes inflated task result sizes (~3 MiB/task, ~60 GiB total
across tasks), leading to spark.driver.maxResultSize failures in terminal
ResultStages (such as RDD checkpointing and file writes) and severe driver
memory pressure.
For more details checkout : https://github.com/apache/spark/issues/58394
*Proposed Solution*
1. Replace the upfront O(numPartitions) SQLMetric accumulators with a single
CollectionAccumulator on StateStoreWriter, allowing each executor task to
append only its own active partition metrics.
2. In StateStoreWriter.getProgress(), the driver aggregates and reports the
top-K instance metrics directly from this single collection accumulator.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
