GC Pressure, Memory, and Watermarks in PySpark Structured Streaming

Your streaming job meets its 30-second SLA in staging, then falls behind by 40 minutes in production. You add executors. It gets worse. The dashboard goes stale. You are not looking at a compute problem — you are looking at a GC, memory, and state problem wearing a compute costume.

This is the operational reality of PySpark Structured Streaming at scale: the math that works on a 5 GB test stream collapses under a 300 GB/hour retail event stream because the JVM was never sized for the object churn a micro-batch creates, the cache strategy deserialized millions of rows into the old generation, and the watermark grew unbounded while late data kept resurrecting expired state.

Here is how to diagnose it, size it, and fix it — with a retail instore dashboarding example running on Dataproc GCP.

🏪 The problem statement: retail instore dashboarding at country scale

You operate a nationwide retail chain with ~2,000 stores. Every store emits POS transactions, inventory movements, and footfall counts to a central streaming pipeline. The business wants a live instore dashboard: “what is selling right now, per store, per category,” refreshed within 30 seconds of the event happening.

The ingestion shape:

The stream joins transaction events against a slowly changing dimension (store master, category taxonomy), aggregates per store_id + category + 1-hour tumbling window, and writes to BigQuery for the dashboard. It also maintains a 30-day rolling state for “store velocity” comparisons.

This is not a toy example. At peak, you are processing 6 GB every 30 seconds, with thousands of keys, state updates, and joins — and your GC pauses are now eating 25–40% of each micro-batch window.

🤔 Questions to seek clarity on before acting

Before touching a single Spark config, answer these. If you cannot answer them with data, you are guessing, and guessing at 30-second latency means breaching the SLA.

1. What is the actual event rate, not the advertised one?

If your peak is 45,000 events/sec but you sized for 8,000, your micro-batch volume is 5.6× larger than planned and the 30-second window will never close.

2. Is the 30-second SLA a processing budget or an end-to-end budget?

T_trigger = 30 s includes:

• Micro-batch planning and task scheduling
• Shuffle read/write
• State store reads/writes (RocksDB on Dataproc local SSD)
• Sink write to BigQuery
• Network between executors and sink

Your processing budget inside Spark must be < 25 s to leave headroom for sink and orchestration. If you budget the full 30 s for Spark processing, you have already lost.

3. What state do you actually need to keep, and for how long?

Watermarking defines how long state survives. If your watermark lags by 10 minutes because of skewed event timestamps, your state store holds 40 minutes of data instead of 30 seconds. Multiply that by 2,000 stores × 50 categories and you have a multi-gigabyte state store that never shrinks.

4. Are you caching anything in the streaming path?

df.persist() inside a foreachBatch or on a streaming DataFrame keeps objects in the JVM heap across micro-batches. Deserialized caching at streaming scale is a GC time bomb. You need to know what is cached, in what format, and whether it is ever evicted.

5. What is your failure mode when latency slips?

Does the dashboard show stale data? Do micro-batches queue up and never catch up? Does the job OOM and restart, losing state? The answer determines whether you tune GC, increase parallelism, or redesign the state logic.

🧠 Why GC kills streaming latency more than batch

In batch, a GC pause is a tax on a finite job. In streaming, a GC pause is a recurring tax on every micro-batch, and if the tax exceeds your trigger interval, you never catch up.

A PySpark executor is a JVM. Every micro-batch creates:

• Row objects for every record read from Kafka
• Hash-map entries for aggregations
• Broadcast lookup structures
• Shuffle buffers and serializer objects
• State store entries and update deltas

When the trigger is 30 seconds and processing takes 22 seconds, an 8-second GC pause means the batch misses its window. The next batch arrives while the previous one is still committing. Backlog accumulates. Latency drifts from 30 seconds to 3 minutes to 40 minutes.

The formula you actually care about:

If t_GC is 12 seconds per 30-second trigger, your effective throughput is 60%. You are paying for 100 cores to get 60 cores of work.

📊 Signal 1: Reading the Spark UI for streaming

Structured Streaming adds a Streaming tab to the Spark UI. Read it in this order:

Streaming    ← START HERE. Batch duration, input rows, processing time
Executors    ← GC Time vs Task Time, per executor
Stages       ← is one task 10x the median? (skew detector)
Jobs         ← which micro-batch stage dominates wall clock?
Storage      ← is your cache spilling? is state store growing?
SQL          ← exchange sizes, state store scan sizes
Environment  ← did your -XX flags even apply?

What to look for in the Streaming tab

The Executors tab ratio

Same as batch: compute GC Time ÷ Task Time per executor. But in streaming, watch it across micro-batches, not cumulatively. An executor that starts at 5% and drifts to 35% over two hours has a memory leak or unbounded state — not a one-time churn event.

Executor | Task Time | GC Time | GC/Task | Trend
1        | 18.2 s    | 0.9 s   | 5%      | stable
2        | 18.1 s    | 1.1 s   | 6%      | stable
3        | 15.4 s    | 5.2 s   | 34%     | RISING — leak or skew
4        | 18.3 s    | 0.8 s   | 4%      | stable

Executor 3 at 34% and rising is your incident. It is either processing a skewed store (one store with 10× the transactions) or accumulating state that is never evicted.

🔍 Signal 2: GC log is ground truth

Turn on executor GC logging. On Dataproc with Spark 3.x / Java 11:

--conf "spark.executor.extraJavaOptions=-Xlog:gc*:time,uptime,level,tags" \
--conf "spark.executor.defaultJavaOptions=-XX:+HeapDumpOnOutOfMemoryError"

Read a single line:

[2026-10-31T10:14:22.418+0000][7.842s][info][gc] GC(214) Pause Young (Normal) (G1 Evacuation Pause) 1841M->612M(4096M) 412.883ms

Classify each pause:

Compute your rates:

In streaming, P(survives young) is driven by state that lives across micro-batches. If your state store holds deserialized rows, every state entry is a candidate for promotion. Serialized state (MEMORY_AND_DISK_SER) keeps state as byte arrays — cheap for the collector to traverse.

🗄️ Memory: caching and persisting in streaming

This is the section that separates production streaming from tutorials.

The trap: MEMORY_AND_DISK in a streaming job

# DANGEROUS in streaming at scale
df.persist(StorageLevel.MEMORY_AND_DISK)

Every cached row is a live JVM object tree: Row, InternalRow, boxed primitives, HashMap entries, Array[Byte] wrappers. At 6 GB micro-batch volume, deserialized caching can inflate to 15–20 GB of live objects. The GC must walk every node on every pass.

The fix: serialized caching

from pyspark import StorageLevel

# Same data as byte arrays: a handful of large objects instead of millions of small ones
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

Serialized, a cached row is bytes in one array. Deserialized, it is an object graph with headers, pointers, and boxed primitives — commonly 2–5× the raw data size, and every node is a live object the collector traverses on every GC pass.

The harder question: do you need to cache at all?

In Structured Streaming, caching a DataFrame inside foreachBatch means the cache survives across micro-batches. If the cached DataFrame is reused once per batch, recomputing it is often cheaper than keeping it alive in the old generation for hours.

Rule: Cache only dimension lookups that change slower than your watermark. Never cache the event stream itself.

🏷️ Watermarking: event time, state, and the metadata explosion

Watermarking is how Structured Streaming bounds state. But it is also where silent state leaks happen.

The retail example

You aggregate sales per store_id + category + 1-hour tumbling window with a 30-second watermark:

from pyspark.sql.functions import col, window, sum as _sum

agg = (
    events
    .withWatermark("event_time", "30 seconds")
    .groupBy(
        col("store_id"),
        col("category"),
        window(col("event_time"), "1 hour")
    )
    .agg(_sum("amount").alias("total_sales"))
)

query = agg.writeStream \
    .outputMode("update") \
    .format("bigquery") \
    .option("table", "project:dataset.sales_agg") \
    .option("checkpointLocation", "gs://your-bucket/checkpoints/") \
    .trigger(processingTime="30 seconds") \
    .start()

query.awaitTermination()

Output mode is not a detail — it is the difference between a live dashboard and a broken one.

For a live low-latency dashboard, outputMode("update") is mandatory. append delays every result by at least one watermark period. complete never lets the state store shrink. update is the only mode that gives you both low latency and bounded state.

What goes wrong

1. Skewed event timestamps: One store’s clock drifts 20 minutes behind. Its events keep arriving after the watermark has passed their window. Structured Streaming cannot drop that state — it must hold it until the watermark catches up. Your state store grows by the drift duration × event rate.

2. Late data beyond watermark: Events arriving after the watermark are dropped silently (for aggregations). The dashboard shows undercounted sales for that store. You either accept it or raise the watermark — which grows state.

3. State store size math:

At peak:

• Distinct keys per batch: 2,000 stores × 50 categories = 100,000
• State per key: ~200 bytes (window aggregate + metadata)
• Watermark lag: 30 seconds = 1 batch
• State entries per batch: 100,000

But if watermark lag drifts to 10 minutes (20 batches):

• State entries: 2,000,000
• State size: ~400 MB per executor, growing

On Dataproc, this state lives in RocksDB on local SSD. The JVM heap holds the state cache. If the cache grows too large, GC pressure spikes and the executor stalls.

The fix

• Set watermark based on actual event-time skew, not hope. Measure max delay from ingestion to processing.
• Monitor state store size in the Spark UI Storage tab. If it grows unbounded, your watermark is too low or your data is skewed.
• Use mapGroupsWithState or flatMapGroupsWithState only with explicit state timeouts. Unbounded stateful operations without timeouts are state leaks by design.

Handling late events without bloating state

Late-arriving POS data can artificially bloat the state store when skewed timestamps drag the watermark backward. The fix is to explicitly drop late events instead of letting them resurrect expired windows:

from pyspark.sql.functions import col, current_timestamp, window, sum as _sum

# Watermark bounds state; anything older is dropped at the source
agg = (
    events
    .withWatermark("event_time", "30 seconds")
    .filter(col("event_time") >= current_timestamp() - expr("INTERVAL 30 MINUTES"))
    .groupBy(
        col("store_id"),
        col("category"),
        window(col("event_time"), "1 hour")
    )
    .agg(_sum("amount").alias("total_sales"))
)

The filter is the guardrail: events older than 30 minutes from processing time are dropped before they reach the state store. Without it, one store with a drifted clock can keep resurrecting windows the watermark has already passed, and your state store grows by drift duration × event rate instead of shrinking.

🪨 RocksDB off-heap memory: the OOM that GC tuning cannot fix

You cannot discuss state store memory without addressing off-heap allocation. RocksDB runs via JNI outside the JVM heap. The garbage collector you tuned so carefully has zero visibility into RocksDB’s memory. If you do not explicitly tune spark.executor.memoryOverhead and enable RocksDB’s bounded memory usage, the executor will be terminated by the Linux OOM killer regardless of how perfectly you tuned the JVM garbage collector.

The failure mode

At peak, your state store holds millions of entries across 2,000 stores × 50 categories. RocksDB allocates memtables, block caches, and bloom filters in off-heap memory. The JVM heap looks healthy in the Spark UI, but the container’s total RSS (heap + off-heap + overhead) exceeds the YARN/Kubernetes memory limit, and the kernel sends SIGKILL. Exit code 137. No Java stack trace. Just a dead executor and a backlog that never recovers.

The fix

--conf spark.executor.memoryOverhead=4g \
--conf spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage=true \
--conf spark.sql.streaming.stateStore.rocksdb.boundedMemoryRatio=0.3

Rule: memoryOverhead must cover RocksDB bounded memory + Python worker memory + netty shuffle buffers + a safety margin. If you see exit code 137 on Dataproc workers, raise memoryOverhead before you touch spark.executor.memory — the heap was never the problem.

Point RocksDB at Local NVMe SSDs, not Persistent Disk

RocksDB writes state to local disk. On GCP Dataproc, if you do not explicitly point the RocksDB state store directory to the attached Local NVMe SSDs, it writes to the standard Persistent Disk (PD). At 6 GB per 30 seconds with 100,000 keys, standard PD IOPS will throttle the job to death — your 30-second SLA breaches from disk I/O, not GC.

On Dataproc, Local SSDs are mounted at /mnt/disks/ssd0/ (or /mnt/1/yarn/ depending on image version). Point the streaming state store checkpoint location there:

spark.conf.set("spark.sql.streaming.stateStore.stateStoreDir", "/mnt/disks/ssd0/spark-streaming-state")

Or via the Dataproc cluster properties:

--properties=spark:spark.sql.streaming.stateStore.stateStoreDir=/mnt/disks/ssd0/spark-streaming-state

Verify the mount exists before submitting the job. Local SSDs are ephemeral — if a worker is preempted or recreated, the directory must be remounted. For streaming state that must survive restarts, back up the checkpoint directory to GCS (gs://your-bucket/state-checkpoints/) and restore on worker replacement.

⚙️ The micro-batch metadata problem: small chunks, big complexity

You said it yourself: too many small chunks in micro-batch create higher complexity on metadata handling than on the data itself.

At 30-second triggers with 45,000 events/sec, each micro-batch is 6 GB. If you partition poorly, you might get 1,000 tiny partitions of 6 MB each. Spark metadata overhead per partition:

• Task scheduling
• Shuffle write/read bookkeeping
• State store opens/closes
• RocksDB compaction triggers
• Serializer context setup

If metadata overhead is 2 seconds per batch and data processing is 4 seconds, 33% of your 30-second window is bookkeeping, not computation.

The math

Example (base): 1.08 GB / 128 MB ≈ 9 partitions Example (peak): 6.08 GB / 128 MB ≈ 48 partitions

If you have 200 executors × 5 cores = 1,000 slots but only 48 partitions, 95% of your cluster sits idle while one executor chokes on a skewed store.

The fix: Repartition by store_id before aggregation to distribute keys evenly. Use spark.sql.shuffle.partitions sized for peak micro-batch volume, not the default 200.

🔒 Backpressure and throttling: capping micro-batch recovery

When a GC pause causes a micro-batch to miss its 30-second window, the next trigger will attempt to pull in the accumulated backlog. At 45,000 events/sec, ingesting even a 3-minute backlog (~8.1M events) will instantly crash the executors with OOMs or shuffle spills that never recover. A production guide must mandate Kafka throttling via maxOffsetsPerTrigger to cap micro-batch sizes during recovery.

The backlog death spiral

Batch N:    GC pause takes 12 s → batch misses 30 s window → backlog = 30 s of events
Batch N+1:  Processes batch N + backlog (60 s of events) → takes 90 s → misses window → backlog = 90 s
Batch N+2:  Processes batch N+1 + backlog (120 s of events) → OOM → executor dies → backlog = ∞

Once you enter this spiral, adding executors makes it worse — more executors pull more offsets, creating larger micro-batches that take longer to process, which creates more backlog.

The fix: maxOffsetsPerTrigger — sized for recovery, not just peak

Cap how many Kafka offsets each micro-batch can consume:

spark.conf.set("spark.sql.streaming.kafka.maxOffsetsPerTrigger", "4050000")

Derive the value:

Example: 45,000 events/sec × 30 s × 3.0 = 4,050,000 offsets per trigger

The recovery multiplier is not optional. If you cap at exactly peak (1×), the system can never burn down a backlog during peak hours. If the pipeline goes down for 5 minutes at lunch, you accumulate 13.5 million events. With a 1.35M cap and 25-second processing, you only burn down 5 seconds of backlog per trigger — recovery takes hours, and your 30-second dashboard SLA is breached the entire time.

Size the cluster for 2–3× peak first. maxOffsetsPerTrigger should match what your hardware can actually process per trigger, not what Kafka is sending. Derive your executor count from the recovery target:

Then set maxExecutors in Dataproc Dynamic Allocation to match that recovery capacity, not just steady-state peak.

Pair this with monitoring on the Streaming tab’s input rows/second. If it flatlines at your maxOffsetsPerTrigger ceiling while consumer lag grows, your processing is too slow — add executors or optimize the query, don’t raise the cap blindly.

🧮 Resource sizing math for Dataproc GCP

Dataproc on GCP: you size worker nodes (executors), not individual cores directly. A Dataproc worker node runs one executor (or multiple, depending on config). Common machine type: n2-standard-16 (16 vCPU, 64 GB RAM) or n2-standard-32 (32 vCPU, 128 GB).

Step 1: Cores needed per trigger

At peak, 48 partitions × 1.5 s/partition = 72 task-seconds.

But shuffle stages need more parallelism:

Step 2: Executors on Dataproc

--executor-cores 4               # 4 cores per executor
--executor-memory 16g            # 16 GB heap per executor
--conf spark.executor.memoryOverhead=4g  # Python + off-heap
--num-executors 6                # 6 × 4 = 24 cores

On Dataproc, 6 n2-standard-16 workers gives you 96 vCPU total, but you only need 24 cores for Spark. The rest is headroom for peak and DRA. Dataproc’s advantage is you can add workers without restarting the job.

Step 3: Dynamic allocation for diurnal load

Retail traffic follows store hours. Base load at night is 1/5th of peak. Static sizing for peak wastes 80% of spend off-peak.

spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "2")
spark.conf.set("spark.dynamicAllocation.maxExecutors", "12")
spark.conf.set("spark.dynamicAllocation.schedulerBacklogTimeout", "10s")
spark.conf.set("spark.dynamicAllocation.sustainedSchedulerBacklogTimeout", "20s")
spark.conf.set("spark.dynamicAllocation.executorIdleTimeout", "90s")

Derive bounds:

Base: ceil(1.08 GB / (128 MB × 4)) = ceil(2.1) = 3 → set 2 (safety floor) Peak: ceil(6.08 GB / (128 MB × 4)) = ceil(11.9) = 12 → set 12

Step 4: Memory split inside the heap

Off-heap memory is invisible to the GC. A 4 GB off-heap state cache means 4 GB less heap the collector must walk every micro-batch.

🛠️ Dataproc GCP specific configuration

gcloud dataproc clusters create streaming-retail \
  --region=us-central1 \
  --master-machine-type=n2-standard-16 \
  --worker-machine-type=n2-standard-16 \
  --num-workers=6 \
  --master-boot-disk-size=500GB \
  --worker-boot-disk-size=500GB \
  --image-version=2.1-debian11 \
  --optional-components=ANACONDA,JUPYTER \
  --properties=core:fs.defaultFS=gs://your-bucket/ \
  --properties=spark:spark.dynamicAllocation.enabled=true \
  --properties=spark:spark.dynamicAllocation.minExecutors=2 \
  --properties=spark:spark.dynamicAllocation.maxExecutors=12 \
  --properties=spark:spark.executor.cores=4 \
  --properties=spark:spark.executor.memory=16g \
  --properties=spark:spark.executor.memoryOverhead=4g \
  --properties=spark:spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage=true \
  --properties=spark:spark.sql.streaming.stateStore.rocksdb.boundedMemoryRatio=0.3 \
  --properties=spark:spark.sql.streaming.kafka.maxOffsetsPerTrigger=1350000 \
  --properties=spark:spark.sql.shuffle.partitions=200 \
  --properties=spark:spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --properties=spark:spark.memory.offHeap.enabled=true \
  --properties=spark:spark.memory.offHeap.size=4g \
  --properties=spark:spark.executor.extraJavaOptions=-Xlog:gc*:time,uptime,level,tags \
  --initialization-actions=gs://dataproc-initialization-actions/python/pip-install.sh

Key Dataproc considerations:

• Local SSD for state store: RocksDB spills to local disk. Use 500 GB+ boot disks or attach local SSDs. State store size grows with watermark lag; monitor disk usage.
• Preemptible workers: Use for base load only. Streaming state is not preemption-safe — do not use preemptibles for workers holding streaming state.
• Regional vs single-zone: Regional clusters survive zone failures but add cross-zone network latency. For 30-second SLA, single-zone with fast local SSD is often better.
• BigQuery sink: Use the Spark BigQuery connector with writeStream.foreachBatch. Batch writes per micro-batch reduce BigQuery API cost and quota pressure.

📦 Sink latency blind spot: BigQuery in the 30-second budget

For a 30-second SLA, writing the final aggregations to BigQuery can consume 5–15 seconds of that budget. If the network or the BigQuery load API throttles, no amount of GC tuning will save the SLA — you will breach the window before a single executor even finishes processing.

The sink consumes the trigger

Your 30-second budget is not all Spark processing time:

If Spark processing takes 20 s and BigQuery takes 12 s, you are at 32 s — already breached. The Streaming tab will show batch duration under 25 s (Spark only), giving you false confidence while the dashboard goes stale.

The fix

1. Budget the sink separately. Treat sink time as a first-class citizen in your SLA math. If BigQuery consistently takes >10 s, reduce write frequency (write every 2–3 micro-batches instead of every trigger).

2. Use foreachBatch with batched writes. Do not write row-by-row. Accumulate micro-batch results and write in larger chunks to amortize BigQuery API overhead:

def write_to_bigquery(batch_df, batch_id):
    batch_df.write \
        .format("bigquery") \
        .option("table", "project:dataset.sales_agg") \
        .mode("append") \
        .save()

agg.writeStream \
    .foreachBatch(write_to_bigquery) \
    .option("checkpointLocation", "gs://your-bucket/checkpoints/") \
    .trigger(processingTime="30 seconds") \
    .start()

3. Monitor sink duration. In the Spark UI Streaming tab, the Sink duration column tells you how long each micro-batch spent writing. If it is creeping toward 10–15 s, your BigQuery write is the bottleneck — not GC, not partitioning, not memory.

4. Consider streaming insert vs batch load. For sub-30-second latency, BigQuery’s streaming insert API has lower latency than batch load but higher cost and stricter quota. Choose based on your SLA vs budget tradeoff.

🔄 The diagnosis loop for streaming

Internalize this loop. Do not skip ahead.

Streaming tab: batch duration vs T_trigger
│
├─ Is batch duration < 25 s consistently?
│   ├─ NO ──► Executors tab: GC/Task ratio per executor
│   │   ├─ One executor high ──► Stages tab: skew on store_id
│   │   │   fix: repartition by store_id, salt hot keys
│   │   └─ All executors high ──► GC log: churn vs retention
│   │       ├─ Many young GC, low reclaim ──► MEMORY_AND_DISK_SER, Kryo
│   │       └─ Pause Full present ──► heap too small or state too large
│   └─ YES ──► Storage tab: state store size bounded?
│       ├─ NO ──► Watermark too low or event-time skew
│       │   fix: raise watermark, monitor late data ratio
│       └─ YES ──► Sink tab: BigQuery write time
│           ├─ High ──► increase foreachBatch batch size, reduce write frequency
│           └─ Stable but total latency > 30 s ──► Sink is the bottleneck, not Spark
│
├─ Exit code 137 / OOM killed on Dataproc workers?
│   └─ YES ──► RocksDB off-heap memory exceeded limit
│       fix: raise spark.executor.memoryOverhead, enable rocksdb.boundedMemoryUsage
│
└─ Consumer lag growing after GC incident?
    └─ YES ──► maxOffsetsPerTrigger too low or processing too slow
        fix: raise cap, add executors, or optimize query — do not raise blindly

One lever per iteration. Change four configs at once and you learn nothing — you get a new latency number with no causal story.

📉 What it looks like on the bill

Numbers below are representative of a real retail streaming pipeline after three changes: repartition by store_id, MEMORY_AND_DISK_SER for dimension cache, and off-heap state store on Dataproc.

The arithmetic: before, 12 workers × 16 vCPU × 24 h = 4,608 core-hours. But you only needed 24 cores at peak. After DRA, average is ~48 core-hours × 24 h = 1,152. ~75% less compute for the same SLA.

You did not process less data. You stopped paying for GC pauses, unbounded state, and idle executors.

✅ The streaming checklist

Run it in this order:

1. Streaming tab — record batch duration vs 30 s trigger. Is it stable or drifting?
2. Executors tab — compute GC Time ÷ Task Time per executor. One high = skew. All high = pressure.
3. Stages tab — compare max task duration to p75 on the dominant streaming stage. Rule out skew before touching JVM.
4. Storage tab — is state store size bounded by watermark? If growing, your watermark is wrong.
5. GC log — confirm -Xlog:gc*:... is on executors (not driver). Count Pause Full occurrences per hour.
6. Cache audit — find every .persist() in the streaming path. Convert to MEMORY_AND_DISK_SER or remove.
7. Repartition check — are keys distributed evenly across executors? If one store is 10× the volume, salt or repartition.
8. Dataproc DRA — set min/max executors from the math above. Monitor scale-up/down frequency.
9. RocksDB off-heap — enable spark.sql.streaming.stateStore.rocksdb.boundedMemoryUsage and size spark.executor.memoryOverhead for the RocksDB working set.
10. Throttling — set spark.sql.streaming.kafka.maxOffsetsPerTrigger to prevent backlog death spiral during recovery.
11. Late data guard — add a filter to drop events older than your watermark tolerance before they reach the state store.
12. Sink duration — monitor BigQuery write time in the Streaming tab; it can consume 5–15 s of your 30-second budget.
13. Re-measure — compare the same two numbers (batch duration, GC/Task ratio) after each change.

#PySpark #StructuredStreaming #GC #Dataproc #GCP #StreamingEngineering #RetailAnalytics #PerformanceTuning #DataEngineering #JVM