Hacktoberfest 2026: những issue maintainer đã đánh dấu cho tháng Mười, đang mở và phù hợp người mới. Xem issue Hacktoberfest

[SQL] State-store instance metrics are shipped in every task result

Đã đóng
  • Có gì không ổn?
#58,394 2 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 2 ngày

@Diveyam-Mishra đang làm issue này rồi.

Từ ngày 2/9/2026.

  • #58476 của @Diveyam-Mishra — đang mở

Đánh giá

Độ khó
5/5
Thời gian dự kiến
Hơn một tuần
Mức phù hợp với người mới
35/100
Loại issue
Lỗi
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
scala, sql
Lĩnh vực
data-engineering, distributed-systems

Hướng nghiên cứu

Bắt đầu với StateStoreWriter.stateStoreInstanceMetricsWithIds và setStoreInstanceMetrics, sau đó lần theo cách xử lý accumulator qua AccumulatorV2.readObject, Task.collectAccumulatorUpdates, Executor và TaskSetManager.canFetchMoreResults. Tái hiện workload ResultStage có số lượng partition cao được mô tả và kiểm tra kích thước kết quả của task. Phần hoàn tất cần ngăn các metric instance của state store không liên quan làm phình to kết quả task đầu cuối, đồng thời vẫn giữ lại các metric tiến độ bắt buộc của state operator.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

Summary

Stateful operators currently create one state-store instance metric accumulator for every combination of:

state partition
x supported provider metric
x state-store name

Every task then receives and registers all of those accumulators, even though it processes only one state partition. On task completion, all external accumulators are included in the task result.

For a job with 20,000 shuffle partitions, RocksDB state, and one default store, this can result in approximately 20,000 state-instance metric updates per task.

Reproduction

A representative workload is:

spark.sql.shuffle.partitions = 20000
stateful operator using RocksDB state
checkpoint or write the stateful result

Observed behavior:

per-task result size: approximately 3 MiB
number of tasks: approximately 20,000
terminal stage: ResultStage

The cumulative task-result data can therefore be approximately 60 GiB. The job may fail with spark.driver.maxResultSize even though the actual rows are written or checkpointed by executors.

The same issue can occur with direct file writes because the final write stage is submitted through SparkContext.runJob and uses ResultTasks. Each task returns a commit result plus accumulator updates.

Root cause

Relevant code paths on master include:

  • StateStoreWriter.stateStoreInstanceMetricsWithIds creates metrics for all state partitions, supported metrics, and store names.
  • StateStoreWriter.setStoreInstanceMetrics updates the corresponding per-instance SQLMetric.
  • AccumulatorV2.readObject registers all deserialized accumulators with the task.
  • Task.collectAccumulatorUpdates sends all external accumulators on successful task completion.
  • Executor serializes those accumulator updates into DirectTaskResult.
  • TaskSetManager.canFetchMoreResults applies spark.driver.maxResultSize to ResultStages.

ignoreIfUnchanged and numStateStoreInstanceMetricsToReport are applied while generating state-operator progress after the driver has received and merged the updates. They do not reduce the task-result payload.

Why checkpointing and direct writes are affected

RDD checkpoint materialization calls SparkContext.runJob, making the checkpointing stage a ResultStage.

Direct file writes also use runJob and return per-task write results or commit messages to the driver. Those task results contain the write result plus accumulator updates, so state-store metrics can contribute to the cumulative result size.

The spark.driver.maxResultSize check is not enforced for upstream ShuffleMapStages, but it is enforced for these terminal ResultStages.

Impact

This behavior causes:

  • inflated task-result serialization and network traffic;
  • increased executor and driver deserialization work;
  • large CompletionEvent payloads on the driver;
  • driver heap pressure if result processing falls behind task completion;
  • possible spark.driver.maxResultSize failures;
  • increased per-task metric metadata retained by scheduler or UI structures.

Scope

This is not specific to TransformWithState. It can affect any stateful operator using StateStoreWriter and a provider that exposes state-store instance metrics.

Multiple state stores, such as those used by streaming joins, multiply the number of metrics included in each task result.

Ngôn ngữ chính
Scala
Star
44.2k
Fork
29.5k
Merge trung bình
8 giờ 43 phút
Pull request đã merge (30 ngày)
1

Chuẩn bị môi trường

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/spark

Tất cả issue của apache/spark

Issue tương tự

Thêm issue về Scala

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.