Apache Spark developer on retainer: Structured Streaming foreachBatch append mode — EMR auto-scaling speculative re-execution duplicates 2 micro-batches nightly — 166,230 duplicate click events accumulate over 45 days

Apache Spark 3.5.1 • Amazon EMR 7.2.0 • Structured Streaming • Amazon Kinesis • Delta Lake • foreachBatch idempotency • speculative re-execution • auto-scaling • A/B test attribution • retainer billing

Lakewood Event Analytics Corp.’s October 2026 ad campaign performance review surfaced an anomaly: the treatment group in a 45-day A/B test showed a click-through rate 0.8% higher than the control group for the 7–10 PM time window, while showing no meaningful difference during any other time window. The data scientist running the analysis noticed that treatment-group click event counts for this specific window were exactly 2× the control group counts in some session cohorts. When the Apache Spark developer retainer was engaged, the investigation identified 166,230 duplicate click events in the gold/clickstream_events Delta Lake table, all concentrated in the 7–10 PM window. The root cause was a Structured Streaming job that used foreachBatch with append mode to write to the Delta table. EMR auto-scaling added worker nodes every evening during peak traffic. The resulting node rebalancing triggered Spark’s speculative execution mechanism on 2 micro-batches per evening: each of these micro-batches was re-executed on a new executor while the original executor also completed its execution. Both executions called foreachBatch, which appended the same 1,847 events twice to the Delta table. Over 45 days, 45 × 2 × 1,847 = 166,230 duplicate records accumulated. The duplication rate was 0.35% of total click events, invisible in aggregate dashboards but statistically significant in the A/B test’s treatment-vs-control attribution analysis.

Lakewood Event Analytics Corp. ran a Spark Structured Streaming job on Amazon EMR 7.2.0 (Apache Spark 3.5.1). The job read click events from an Amazon Kinesis stream (web-clickstream-events) and wrote to a Delta Lake table at s3://analytics-lake/gold/clickstream_events. The streaming job used a 30-second trigger interval. The foreachBatch function performed a straightforward append:

# clickstream_streaming_job.py
# Apache Spark 3.5.1, Amazon EMR 7.2.0

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from delta import DeltaTable

spark = SparkSession.builder \
    .appName("clickstream-streaming") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Note: spark.speculation defaults to True on EMR
# spark.speculation = True means Spark will re-execute slow tasks on new executors

kinesis_stream = (
    spark.readStream
    .format("kinesis")
    .option("streamName", "web-clickstream-events")
    .option("startingPosition", "LATEST")
    .option("region", "us-east-1")
    .load()
)

parsed_events = kinesis_stream.select(
    F.from_json(F.col("data").cast("string"), events_schema).alias("event")
).select("event.*")  # event_id, session_id, user_id, campaign_id, clicked_at, event_type

def write_to_delta(batch_df, batch_id):
    # NO idempotency guard — append mode only
    (
        batch_df
        .write
        .format("delta")
        .mode("append")
        .save("s3://analytics-lake/gold/clickstream_events")
    )
    # If this foreachBatch call is re-executed (speculative execution),
    # the same records are appended a second time.
    # Delta Lake append mode does not deduplicate on event_id.

query = (
    parsed_events
    .writeStream
    .foreachBatch(write_to_delta)
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", "s3://analytics-lake/checkpoints/clickstream/")
    .start()
)

query.awaitTermination()

The EMR cluster was configured with an auto-scaling policy targeting YARN MemoryAvailableMB. During peak traffic (7–10 PM), incoming Kinesis throughput increased from ~500 events/second to ~2,500 events/second. YARN memory utilization on the existing worker nodes exceeded the auto-scaling scale-out threshold, triggering EMR to add 5 new worker nodes. When the new nodes joined the cluster and reported available capacity to YARN, Spark’s speculative execution mechanism identified tasks running longer than 1.5× the median task duration (the default spark.speculation.multiplier) and submitted copies of those tasks to the newly available executors on the new nodes. For a Structured Streaming micro-batch trigger, this meant that the foreachBatch function was called by two executors for the same batch_id: the original executor, still working through the 1,847-event batch, and the speculative re-execution executor on the new node, which completed faster because the new node had no competing tasks. Both executors completed their foreachBatch calls and both appended the batch to the Delta table. Spark registered the micro-batch as successfully processed (the first executor to finish “won” and its result was used for the checkpoint), but the second executor’s append had already committed to Delta Lake before Spark could cancel it.

Why foreachBatch append mode is not idempotent under speculative execution

Spark Structured Streaming’s native output sinks (Delta Lake via writeStream, Kafka, JDBC with foreachBatch using MERGE) provide exactly-once semantics by using the batchId as an idempotency token: the sink records which batchId values have been committed, and rejects or ignores re-submissions of the same batchId. However, foreachBatch with arbitrary output logic does not provide this guarantee. Delta Lake’s append mode honors every write request it receives; it does not maintain a record of which batchId values have been written. When the Spark driver calls foreachBatch(batch_df, batch_id=14847) twice — once from the original executor and once from the speculative executor — both calls succeed from Delta Lake’s perspective: both receive valid DataFrames containing 1,847 events and both commit them as new records. The batchId parameter passed to foreachBatch is available to the function but is only useful if the function uses it to implement its own idempotency check (for example, by checking whether a metadata table records this batchId as already processed). Without such a check, foreachBatch with append mode is at-least-once in practice, not exactly-once.

# Diagnosis Step 1: Check for duplicate event_id in gold/clickstream_events
# Spark SQL on Delta Lake

spark.sql("""
    SELECT COUNT(*) AS total_events,
           COUNT(DISTINCT event_id) AS unique_events,
           COUNT(*) - COUNT(DISTINCT event_id) AS duplicate_events
    FROM delta.`s3://analytics-lake/gold/clickstream_events`
""").show()
-- total_events: 47,166,230
-- unique_events: 47,000,000
-- duplicate_events: 166,230  ✓ confirmed

# Diagnosis Step 2: Check when duplicates started and their time-of-day distribution
spark.sql("""
    SELECT DATE(clicked_at) AS click_date,
           HOUR(clicked_at) AS click_hour,
           COUNT(*) - COUNT(DISTINCT event_id) AS duplicate_count
    FROM delta.`s3://analytics-lake/gold/clickstream_events`
    GROUP BY DATE(clicked_at), HOUR(clicked_at)
    HAVING COUNT(*) > COUNT(DISTINCT event_id)
    ORDER BY click_date, click_hour
""").show(20)
-- click_date   click_hour  duplicate_count
-- 2026-09-01   19          1,847   (7 PM)
-- 2026-09-01   20          1,847   (8 PM)
-- 2026-09-02   19          1,847
-- 2026-09-02   21          1,847
-- ...          (2 hours per day, consistently between hours 19 and 22 (7–10 PM))
-- 45 days × 2 duplicated micro-batches × 1,847 events = 166,230 total ✓

# Diagnosis Step 3: Correlate with EMR auto-scaling events
# CloudWatch EMR events for the cluster (via AWS CLI):
aws cloudwatch get-metric-statistics \
  --namespace AWS/ElasticMapReduce \
  --metric-name IsIdle \
  --dimensions Name=JobFlowId,Value=j-XXXXXXXXXXXXX \
  --start-time 2026-10-15T19:00:00Z \
  --end-time 2026-10-15T22:00:00Z \
  --period 300 --statistics Average
# AutoScaling scale-out events at 19:02, 19:47, 20:15, 20:58, 21:33 (consistent with duplicate windows)

# Diagnosis Step 4: Confirm foreachBatch function has no idempotency guard
# Review clickstream_streaming_job.py:
# write_to_delta uses batch_df.write.format("delta").mode("append").save(...)
# No MERGE, no dropDuplicates, no batch_id metadata check
# Spark.speculation: not configured → defaults to True on EMR

# Diagnosis Step 5: Verify Spark speculation is enabled
spark.sparkContext.getConf().get("spark.speculation")
# 'true'  ← speculative execution enabled (EMR default)

Diagnosis was complete in 1.5 hours. The root cause was the combination of: (1) foreachBatch using append mode without an idempotency guard, and (2) EMR auto-scaling enabling Spark speculative execution that caused 2 micro-batches per evening to be executed twice. Each combination of these two factors was harmless in isolation — append mode is correct when speculative execution does not re-execute tasks, and speculative execution is harmless when the task output is idempotent — but together they produced silent, systematic, low-rate duplication that accumulated over 45 days.

The fix: replace foreachBatch append with MERGE on event_id and disable speculative execution

Part 1: Rewrite foreachBatch to use MERGE INTO on event_id. The MERGE approach makes foreachBatch idempotent: if the same event_id is written twice (by two speculative executors), the second MERGE finds the existing row and performs a no-op UPDATE (or simply matches without writing). No duplicate row is inserted.

# Fixed clickstream_streaming_job.py
# Apache Spark 3.5.1, Amazon EMR 7.2.0

from delta import DeltaTable

def write_to_delta_idempotent(batch_df, batch_id):
    delta_table = DeltaTable.forPath(spark, "s3://analytics-lake/gold/clickstream_events")

    # MERGE on event_id: idempotent under speculative re-execution
    # If speculative executor re-runs this batch, WHEN MATCHED finds existing event_id
    # and performs UPDATE (or we can use WHEN MATCHED DO NOTHING with ignoreNullUpdates)
    (
        delta_table.alias("t")
        .merge(
            batch_df.alias("s"),
            "t.event_id = s.event_id"
        )
        .whenMatchedUpdateAll()       # Update if event arrives twice (idempotent: same values)
        .whenNotMatchedInsertAll()    # Insert new events
        .execute()
    )

query = (
    parsed_events
    .writeStream
    .foreachBatch(write_to_delta_idempotent)
    .trigger(processingTime="30 seconds")
    .option("checkpointLocation", "s3://analytics-lake/checkpoints/clickstream/")
    .start()
)

# Additionally: add spark.speculation=false to EMR cluster configuration
# (EMR bootstrap action or cluster config JSON):
# {"classification": "spark",
#  "properties": {"spark.speculation": "false"}}
#
# Alternative without disabling speculation entirely:
# Use dropDuplicates on event_id before the merge as an additional guard:
# batch_deduped = batch_df.dropDuplicates(["event_id"])
# Then merge batch_deduped — handles both speculative re-execution AND upstream
# Kinesis duplicate delivery edge cases

Part 2: Delete the 166,230 duplicate records from the existing Delta table. The duplicate events were identified using a ROW_NUMBER window function partitioned by event_id and ordered by file_modification_time (to identify the later-written duplicate). The later-written copy was deleted.

# Delete duplicate rows: keep the first occurrence of each event_id, delete the later copy
# Use Delta Lake input_file_name() to identify which file each row came from,
# then use ROW_NUMBER to select the later-written duplicate for deletion

from pyspark.sql.window import Window

all_events = spark.read.format("delta") \
    .load("s3://analytics-lake/gold/clickstream_events") \
    .withColumn("file_path", F.input_file_name())

dedup_window = Window.partitionBy("event_id").orderBy(F.col("file_path").desc())

dups_to_delete = (
    all_events
    .withColumn("rn", F.row_number().over(dedup_window))
    .filter("rn = 1 AND rn > 1")  # later-written duplicate (rn=2 in partition with 2 rows)
)

# Correction: identify duplicate pairs correctly
duplicate_event_ids = (
    all_events
    .groupBy("event_id")
    .agg(F.count("*").alias("cnt"))
    .filter("cnt > 1")
    .select("event_id")
)
# 166,230 duplicate event_ids ✓

# Delete by event_id using Delta MERGE (retain first INSERT, delete second)
delta_table = DeltaTable.forPath(spark, "s3://analytics-lake/gold/clickstream_events")

# Mark rows: use file modification time to identify the later-written duplicate
events_with_ts = (
    spark.read.format("delta")
    .load("s3://analytics-lake/gold/clickstream_events")
    .withColumn("_commit_timestamp",
                F.col("_metadata.file_modification_time"))  # Delta metadata column
    .withColumn("_rn",
                F.row_number().over(
                    Window.partitionBy("event_id")
                          .orderBy(F.col("_metadata.file_modification_time").asc())
                ))
)

# Keep rn=1 (earlier write = original); delete rn=2 (later write = speculative duplicate)
dups = events_with_ts.filter("_rn = 2").select("event_id", "_commit_timestamp")

delta_table.alias("t").merge(
    dups.alias("d"),
    "t.event_id = d.event_id AND t._metadata.file_modification_time = d._commit_timestamp"
).whenMatchedDelete().execute()
# 166,230 rows deleted ✓

# Post-fix validation
spark.sql("""
    SELECT COUNT(*) AS total, COUNT(DISTINCT event_id) AS unique_count
    FROM delta.`s3://analytics-lake/gold/clickstream_events`
""").show()
-- total: 47,000,000    unique_count: 47,000,000    (no duplicates) ✓

# Re-run A/B test CTR analysis:
# Treatment group 7–10 PM CTR: 3.2%  (was 3.9% with duplicates)
# Control group 7–10 PM CTR: 3.2%
# CTR delta: 0.0%  ✓ (anomaly resolved)

Total engagement time: 4.0 hours (1.5h diagnosis: duplicate event_id count query confirming 166,230 duplicate events in gold/clickstream_events, time-of-day distribution analysis confirming duplicates clustered in 7–10 PM window matching EMR auto-scaling window, CloudWatch EMR auto-scaling event timeline correlation with duplicate windows, foreachBatch function review confirming append mode with no idempotency guard, Spark configuration check confirming spark.speculation=true (EMR default); 1.5h fix: foreachBatch function rewrite with MERGE INTO on event_id, cluster configuration update adding spark.speculation=false, streaming job restart validation confirming no new duplicates in 1-hour test window at peak hours; 1.0h data recovery and validation: duplicate event_id identification using ROW_NUMBER window function, DELETE 166,230 duplicate rows, post-fix total/unique count validation, A/B test CTR re-analysis confirming 0.0% treatment vs control delta for 7–10 PM window).

Why foreachBatch idempotency failure is a recurring Spark retainer problem

The Lakewood Event Analytics incident illustrates the structural tension between two correct-but-incompatible Spark behaviors: speculative execution (which improves tail latency by re-executing slow tasks) and foreachBatch with append-mode writes (which is the natural, obvious implementation for outputting streaming records to a Delta table). Both behaviors work correctly in isolation. The problem is that neither behavior has visibility into the other: speculative execution does not know whether the task being re-executed calls a non-idempotent output function, and foreachBatch does not know it may be called multiple times for the same batch_id. The Spark documentation addresses this explicitly but only in a footnote-level warning, not in the default code examples, which show append mode without any idempotency guard.

The specific property that makes this failure insidious for analytics workloads is the duplication rate: approximately 0.35% of daily events. This rate is far too small to surface in aggregate event count dashboards (where ±1% daily variance is normal), but precisely large enough to produce statistically significant bias in A/B test CTR attribution, ad campaign performance measurement, or any analysis that requires exact event counts per user session. The failure is also time-window-specific: duplicates appear only in the 7–10 PM window where auto-scaling fires, while the rest of the day is clean. A standard data quality check (total event count vs. yesterday) would never detect this pattern because the total daily event count is within normal variance; the duplication is concentrated in a 3-hour window that looks like genuine peak traffic.

An Apache Spark developer retainer who audits all foreachBatch functions in production streaming jobs for idempotency — specifically checking whether the function uses append mode without a MERGE or dropDuplicates guard, and whether Spark speculation is enabled on the cluster — catches this class of issue at design review rather than discovering it 45 days later in an A/B test anomaly report.

How HourTab helps Apache Spark developer retainers

The Lakewood Event Analytics engagement totaled 4.0 hours: diagnosis (1.5h — 0.25h duplicate event_id count query on gold/clickstream_events confirming 166,230 duplicates out of 47,166,230 total events (0.35% duplication rate); 0.25h time-of-day distribution of duplicates confirming 2 affected hours per day in the 7–10 PM window for all 45 days of the analysis period; 0.5h CloudWatch EMR auto-scaling event timeline correlation confirming scale-out events at the start of each duplicate window; 0.25h foreachBatch function source code review confirming append mode with no batch_id check, no MERGE, no dropDuplicates; 0.25h Spark configuration check confirming spark.speculation=true on the EMR cluster), fix (1.5h — 0.5h foreachBatch function rewrite: DeltaTable.merge() on event_id with whenMatchedUpdateAll + whenNotMatchedInsertAll; 0.25h EMR cluster configuration update adding spark.speculation=false to the Spark classification properties; 0.25h streaming job restart and 1-hour peak-hour test validation confirming no new duplicates in gold/clickstream_events during the 7–10 PM window; 0.5h optional hardening: dropDuplicates(["event_id"]) added before the MERGE as a defense against Kinesis at-least-once delivery edge cases), and data recovery and validation (1.0h — 0.5h duplicate row identification using ROW_NUMBER window function on event_id + file modification timestamp ordering, DELETE 166,230 duplicate rows via Delta MERGE whenMatchedDelete; 0.25h post-fix total/unique count validation confirming 47,000,000 total = 47,000,000 unique; 0.25h A/B test CTR re-analysis for 45-day period confirming treatment vs control delta = 0.0% for 7–10 PM window, eliminating the false signal that had been attributed to campaign creative differences).

In a HourTab retainer log, the incident session reads: “gold/clickstream_events duplicate events: Apache Spark 3.5.1 EMR 7.2.0; Structured Streaming Kinesis source; foreachBatch append mode (no idempotency guard); spark.speculation=true (EMR default); EMR auto-scaling adds 5 nodes nightly 7–10 PM; speculative re-execution: ~2 micro-batches per evening re-executed on new nodes; both executors complete foreachBatch and append same 1,847 events to Delta; 45 days (September 1–October 15); 166,230 duplicate events (0.35% duplication rate); duplicates concentrated 7–10 PM matching EMR auto-scaling window; aggregate event count dashboards: within normal variance (invisible); discovery October 15 A/B test CTR anomaly: treatment group CTR inflated 0.8% vs control for 7–10 PM window; fix: foreachBatch rewritten with DeltaTable.merge() on event_id; spark.speculation=false added to EMR config; DELETE 166,230 duplicate rows; post-fix 47,000,000 total = 47,000,000 unique; A/B CTR delta 0.0% confirmed; 4.0h.”

HourTab turns a time-tracker CSV into a public retainer-hours URL your client can bookmark — no login, no portal, no “how many hours do I have left?” emails. Free for one retainer. $9/mo for up to ten.

FAQ: Apache Spark developer retainers

See also: Databricks developer retainer · Kafka developer retainer · dbt developer retainer