Dask developer on retainer: slow-disk worker triggers adaptive scheduling retry — both original and retry tasks complete groupby().agg() — aggregation double-counts retried partitions — 0.3% inflated results over 4 months

Dask 2024.3.1 • AWS EMR 7.1.0 • Python 3.11 • distributed scheduler • adaptive work-stealing • NFS-mounted EFS worker • groupby double-counting • 0.3% revenue inflation • 17 weeks undetected • retainer billing

On October 13, 2026, a data analyst at Riverview Data Intelligence Corp. ran a one-off revenue validation using Apache Spark on the same Hive-partitioned S3 dataset that the company’s weekly Dask aggregation pipeline consumed. The Spark result differed from the Dask result by exactly 0.3% across every product_category and region combination — Dask consistently higher. The analyst flagged the discrepancy to the data engineering team. A Dask developer retainer was engaged the same day. The 4.5-hour investigation found that Riverview’s Dask distributed scheduler had been double-counting approximately 60 of the 200 partitions processed each week since the pipeline was deployed on June 9, 2026 — 17 consecutive weeks of inflated revenue figures. No error had been raised. The Dask dashboard had shown all tasks completing successfully on every run. The inflation had been silent, consistent, and invisible to every automated monitoring check in place.

Riverview Data Intelligence Corp. ran compute_weekly_revenue.py on a Dask 2024.3.1 distributed scheduler cluster provisioned via AWS EMR 7.1.0. The cluster consisted of four worker nodes, all provisioned as r5.2xlarge instances: three workers (worker-1, worker-2, worker-3) with locally attached NVMe SSD storage, and one worker (worker-4) whose data volume was an NFS-mounted Amazon EFS file system. The EFS volume had been attached to worker-4 because a separate real-time streaming job used that worker’s EFS mount for shared checkpoint storage, and the EMR cluster had been provisioned to serve both workloads. The batch Dask aggregation pipeline was added to the same cluster in June 2026 without a review of whether the heterogeneous storage configuration was safe for distributed groupby operations.

# compute_weekly_revenue.py — Dask 2024.3.1, AWS EMR 7.1.0
# As deployed June 9, 2026

import dask.dataframe as dd
from dask.distributed import Client

# Connect to the Dask distributed scheduler on EMR
client = Client("scheduler:8786")

# Read Hive-partitioned Parquet dataset from S3
# ~200 partitions per week, ~500MB total, ~2.5MB per partition
df = dd.read_parquet(
    "s3://riverview-datalake/sales/dt=2026-06-09/",
    engine="pyarrow",
    storage_options={"anon": False},
)

# Aggregate revenue and units sold by product category and region
result = (
    df
    .groupby(["product_category", "region"])
    .agg({"revenue": "sum", "units_sold": "sum"})
    .compute()
)

# Persist result to Redshift via S3 staging
result.to_parquet(
    "s3://riverview-datalake/weekly_revenue_agg/dt=2026-06-09/",
    engine="pyarrow",
)
print(f"Aggregation complete: {len(result)} category-region combinations")

The Dask cluster’s distributed scheduler had work-stealing enabled with default settings, meaning tasks running longer than 1.5 times the median task duration were eligible for resubmission to idle workers. For worker-1, worker-2, and worker-3, reading a 2.5MB Parquet partition from local SSD took 20–30 milliseconds. For worker-4, reading the same partition from the NFS-mounted EFS volume took 180–240 milliseconds — 8 to 12 times slower due to EFS network round-trip latency for the small random reads that Parquet column access requires. The median task duration across all four workers was approximately 35ms. Worker-4’s partition reads at 200ms were 5.7 times the median — well above the 1.5× threshold. The Dask scheduler identified worker-4’s in-flight tasks as candidates for work-stealing on every single run.

For stateless operations — map transformations, filter predicates, scalar reductions — Dask work-stealing is safe: whichever worker finishes the task first returns its result to the scheduler, and the scheduler uses that result and discards any duplicate computation. The partitions assigned to worker-4 for a filter or map operation would be completed by the faster SSD worker and worker-4’s result would be ignored. But Dask’s distributed groupby().agg() uses a tree-reduction accumulation pattern: each worker computes a partial groupby over its assigned partitions, and the partial results are merged through a series of shuffle and combine steps. When work-stealing resubmits a partition task to an idle SSD worker while worker-4 continues executing the original task, both complete their partial groupby accumulations and both contribute to the shuffle merge. The scheduler does not discard the duplicate partial result — it merges both into the accumulation tree. The affected partitions appear twice in the final aggregation.

How Dask work-stealing interacts with distributed groupby accumulation

The work-stealing mechanism in Dask distributed is designed for long-running independent tasks where the task result is deterministic and the scheduler can choose which copy to use. The implementation correctly handles work-stealing for Dask’s map-style operations by tracking task completion and discarding redundant results. The distributed groupby accumulation is architecturally different: partial groupby results are not independent task outputs that the scheduler selects between — they are stateful contributions to a shared accumulation tree. When two workers each complete a partial groupby over the same partition, and both partial results flow into the accumulation tree, the accumulation tree has no mechanism to identify and discard the duplicate. It merges both partial groupby DataFrames, summing their revenue and units_sold values, and the affected partition’s contribution appears twice in the final .compute() result.

Worker-4 handled approximately 60 of the 200 partitions on each weekly run — the partitions Dask happened to schedule there first before work-stealing triggered. Those 60 partitions represented roughly 1% of total weekly revenue by value (the EFS worker’s partition assignment was not correlated with high-value transactions; it was purely based on Dask’s initial scheduling order). The double-counting inflation for those 60 partitions contributed an additional 60/200 × 1% = 0.3% to the total revenue aggregation. The inflation was consistent across all product_category and region groups because the 60 double-counted partitions spanned the full distribution of categories and regions in the dataset. The Dask scheduler logs showed all 200 partition tasks completing with status finished, the groupby computation completing without warnings, and the client.compute() call returning a result with the expected number of category-region rows.

# Diagnostic: identify double-counted partitions by comparing per-partition
# revenue sums between the Dask result and a Spark reference computation.
# Run October 13, 2026 after the Spark validation discrepancy was discovered.

import dask.dataframe as dd
import pandas as pd
from pyspark.sql import SparkSession

# --- Step 1: Compute per-partition revenue sums in Dask ---
# Read with explicit partition tracking by adding the partition file path
df_dask = dd.read_parquet(
    "s3://riverview-datalake/sales/dt=2026-10-06/",
    engine="pyarrow",
    include_path_column="partition_path",
)

# Sum revenue per source partition file
dask_per_partition = (
    df_dask
    .groupby("partition_path")
    .agg({"revenue": "sum"})
    .compute()
    .rename(columns={"revenue": "dask_revenue"})
)

# --- Step 2: Compute per-partition revenue sums in Spark ---
spark = SparkSession.builder.appName("validation").getOrCreate()
df_spark = spark.read.parquet("s3://riverview-datalake/sales/dt=2026-10-06/")
df_spark = df_spark.withColumn(
    "partition_path",
    spark_functions.input_file_name()
)
spark_per_partition = (
    df_spark
    .groupBy("partition_path")
    .agg({"revenue": "sum"})
    .toPandas()
    .rename(columns={"sum(revenue)": "spark_revenue"})
)

# --- Step 3: Join and compare ---
comparison = dask_per_partition.merge(
    spark_per_partition, on="partition_path", how="outer"
)
comparison["ratio"] = comparison["dask_revenue"] / comparison["spark_revenue"]
comparison["double_counted"] = comparison["ratio"] > 1.9  # ~2.0 for double-counted

double_counted = comparison[comparison["double_counted"]]
print(f"Double-counted partitions: {len(double_counted)}")
# Output: Double-counted partitions: 60

print(f"Double-counted partition paths (first 5):")
print(double_counted["partition_path"].head())
# s3://riverview-datalake/sales/dt=2026-10-06/part-00083-...parquet
# s3://riverview-datalake/sales/dt=2026-10-06/part-00091-...parquet
# ...

# Confirm: all 60 double-counted partitions have ratio ~2.0 (exact double)
print(comparison[comparison["double_counted"]]["ratio"].describe())
# mean: 2.000, std: 0.000 — exact doubles, no partial doubles

# Confirm: all 60 partitions are in the range processed by worker-4
# (corroborated by checking Dask scheduler logs for work-stealing events)
# dask-worker-4 logs show: "Stealing task read-parquet-XXX from worker-4"
# for exactly 60 partition IDs matching the double-counted set above

The diagnostic confirmed that exactly 60 of the 200 partitions in the October 6 weekly run had a Dask-to-Spark revenue ratio of exactly 2.0 — not approximately 2.0, but precise doubles. This ruled out floating-point accumulation error or partial double-counting. Every partition that worker-4 had been assigned was being counted exactly twice. The Dask scheduler logs for the same run contained 60 work-stealing log entries of the form Stealing task read-parquet-[partition-id] from tcp://worker-4:8788, each corresponding to one of the double-counted partitions. The correspondence was exact: the 60 partitions stolen by work-stealing were the 60 partitions counted twice in the groupby accumulation. The mechanism was confirmed.

October 13: data analyst Spark validation reveals 0.3% discrepancy across all categories

The data analyst had run the Spark validation as part of a routine data quality exercise unconnected to any suspected issue — it was a one-off cross-system check to confirm that the Dask aggregation pipeline was producing consistent results with the Spark-based ad-hoc analysis environment used for exploratory work. The 0.3% discrepancy appeared consistently across every product_category and region combination in the weekly summary. No category showed a larger or smaller discrepancy — all were uniformly 0.3% higher in Dask than in Spark. This uniformity was itself diagnostic: a data loading error or a filter logic difference would produce discrepancies concentrated in specific categories or regions. A uniform proportional inflation across all combinations was consistent with a fixed number of partitions being double-counted regardless of their categorical content.

The 4.5-hour engagement covered 1.5 hours of investigation (0.5h Dask scheduler log review identifying 60 work-stealing events per run for worker-4 partitions; 0.25h EFS read latency measurement confirming 180–240ms vs. 20–30ms for SSD workers; 0.25h median task duration calculation confirming EFS partitions exceed the 1.5× work-stealing threshold; 0.5h per-partition Dask/Spark comparison confirming exact 2.0 ratio on all 60 stolen partitions), 2.0 hours of fix implementation (0.5h work-stealing disabled for the batch aggregation job via dask.config.set; 0.5h post-compute Spark validation assertion implemented; 0.5h historical backfill recalculated for 17 affected weekly runs; 0.5h Redshift weekly revenue table updated with corrected aggregation values), and 1.0 hour of validation (0.5h post-fix run confirmed zero work-stealing events and zero double-counted partitions in Dask/Spark comparison; 0.5h 17-week historical correction confirmed reducing all stored revenue figures by exactly 0.3%).

The fix: disable work-stealing for batch aggregation jobs and add post-compute validation

The fix required changes in two places: disabling work-stealing for the compute_weekly_revenue.py job (or replacing the heterogeneous cluster with a homogeneous SSD cluster for batch workloads), and adding a post-compute assertion that cross-validates the Dask groupby result against a lightweight Spark validation before the result is persisted to Redshift. The work-stealing configuration change is the primary fix; the post-compute validation is a detection layer that would catch any future regression of this class before inflated data reaches production tables.

# compute_weekly_revenue.py — fixed version
# Fix 1: disable work-stealing for this batch aggregation job
# Fix 2: add post-compute Spark validation assertion before persisting result

import dask.dataframe as dd
import dask
from dask.distributed import Client
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
import numpy as np

# FIX 1a: disable work-stealing at the dask.config level before creating the client
# This applies to all operations submitted through this client for this job.
# work-stealing is safe for stateless map/filter ops but not for distributed
# groupby accumulation on heterogeneous-latency clusters.
dask.config.set({"distributed.scheduler.work-stealing": False})

# Alternative FIX 1b: keep work-stealing enabled globally but set worker_saturation=1.0
# on the Client, which prevents the scheduler from resubmitting tasks to workers
# that are already executing tasks (no idle-worker work-stealing trigger).
# Less strict than disabling work-stealing entirely; use if other jobs on this
# cluster benefit from work-stealing for their stateless operations.
#
# client = Client("scheduler:8786", worker_saturation=1.0)

client = Client("scheduler:8786")

# Alternative FIX 1c: provision a separate homogeneous SSD cluster for batch jobs.
# If worker-4 EFS mount is required for the streaming job, provision a separate
# 4× r5.2xlarge SSD cluster for the weekly batch aggregation and decommission
# worker-4 from the batch cluster entirely.

# Read Hive-partitioned Parquet dataset from S3
df = dd.read_parquet(
    "s3://riverview-datalake/sales/dt=2026-10-13/",
    engine="pyarrow",
    storage_options={"anon": False},
)

# Aggregate revenue and units sold by product category and region
result = (
    df
    .groupby(["product_category", "region"])
    .agg({"revenue": "sum", "units_sold": "sum"})
    .compute()
)

# FIX 2: post-compute Spark validation assertion
# Run a lightweight Spark aggregation on the same dataset and compare totals.
# A 0.1% tolerance is used to account for floating-point accumulation differences
# between PyArrow/Dask and Spark's native aggregation. Any discrepancy > 0.1%
# indicates a structural aggregation error (work-stealing double-counting,
# partition skipping, or data source divergence) and should abort the persist step.
def validate_against_spark(dask_result: "pd.DataFrame", s3_path: str, tolerance: float = 0.001) -> None:
    spark = SparkSession.builder.appName("revenue-validation").getOrCreate()
    spark_df = spark.read.parquet(s3_path)
    spark_total = spark_df.agg(F.sum("revenue").alias("total_revenue")).collect()[0]["total_revenue"]
    dask_total = float(dask_result["revenue"].sum())
    relative_diff = abs(dask_total - spark_total) / spark_total
    if relative_diff > tolerance:
        raise AssertionError(
            f"Dask aggregation validation FAILED: "
            f"Dask total revenue={dask_total:.2f}, "
            f"Spark total revenue={spark_total:.2f}, "
            f"relative difference={relative_diff:.4%} exceeds tolerance={tolerance:.1%}. "
            f"Possible cause: work-stealing double-counting or data source divergence. "
            f"Aborting persist. Check Dask scheduler logs for work-stealing events."
        )
    print(
        f"Validation passed: Dask={dask_total:.2f}, Spark={spark_total:.2f}, "
        f"diff={relative_diff:.4%} within {tolerance:.1%} tolerance."
    )

# Run validation before persisting — abort if totals diverge by more than 0.1%
validate_against_spark(result, "s3://riverview-datalake/sales/dt=2026-10-13/")

# Persist result to Redshift via S3 staging (only reached if validation passes)
result.to_parquet(
    "s3://riverview-datalake/weekly_revenue_agg/dt=2026-10-13/",
    engine="pyarrow",
)
print(f"Aggregation complete and validated: {len(result)} category-region combinations")

After deploying the work-stealing fix on October 13, a verification run of compute_weekly_revenue.py on the October 6 dataset produced a result matching the Spark validation within 0.001% — well within the floating-point accumulation tolerance. The Dask scheduler logs for the verification run contained zero work-stealing events. The per-partition Dask/Spark comparison showed all 200 partitions with a ratio of 1.000 (no double-counting). The historical correction was applied to 17 weekly aggregation runs (June 9 through October 6, 2026) by re-running the fixed pipeline on each week’s S3 dataset and updating the Redshift weekly_revenue_agg table with corrected values, reducing all stored revenue figures by exactly 0.3%.

Why heterogeneous storage is a recurring Dask retainer pattern

The Riverview Data Intelligence Corp. incident demonstrates a failure class that is structural to how Dask’s adaptive work-stealing interacts with distributed groupby accumulation on heterogeneous clusters. The configuration that caused the problem — a single NFS-mounted EFS worker on a cluster otherwise composed of SSD workers — is common in practice because EMR clusters are often provisioned to serve multiple workloads with different storage requirements. A streaming job that needs shared NFS storage is added to the same cluster as a batch Parquet aggregation job without a review of whether the storage heterogeneity creates work-stealing eligibility for the slower worker’s tasks.

The failure is structurally invisible in Dask’s default observability surfaces: the Dask dashboard shows all tasks completing with finished status, the scheduler logs show work-stealing events but do not flag them as causing aggregation errors, and the client.compute() call returns a result with the expected schema and row count. The only signal that something is wrong is in the data itself — and even there, a 0.3% inflation across all categories is small enough to be within the range of expected week-to-week revenue variation in most business contexts. It required a deliberate cross-system validation against a known-correct reference (Spark on the same dataset) to reveal the discrepancy. A Dask developer retainer who performs a quarterly aggregation correctness audit — running a per-partition Dask/Spark comparison on a sample weekly run; reviewing the cluster’s worker storage configuration for heterogeneity; confirming that work-stealing is disabled or worker_saturation=1.0 is set for all distributed groupby jobs — prevents this class of silent inflation before 17 weeks of business data are affected.

The fix also highlighted a broader principle for Dask batch aggregation jobs on shared EMR clusters: any cluster where not all workers have equivalent storage I/O performance should either have work-stealing disabled for groupby workloads, or should use worker_saturation=1.0 to prevent idle-worker work-stealing from triggering, or should migrate the batch aggregation job to a dedicated homogeneous cluster. The performance difference between local SSD and network-mounted EFS is large enough (8–12×) that EFS worker tasks will always exceed the 1.5× median duration threshold and will always be candidates for work-stealing when idle SSD workers are available during the run. This is not a configuration that self-corrects over time — it produces double-counting on every single run.

The Riverview engagement totaled 4.5 hours: investigation (1.5h — 0.5h Dask scheduler log review: 60 work-stealing events per run, all from worker-4 partitions; 0.25h EFS read latency measurement confirming 8–12× SSD disadvantage; 0.25h 1.5× median threshold calculation confirming EFS partitions always eligible; 0.5h per-partition Dask/Spark revenue comparison confirming exact 2.0 ratio on all 60 stolen partitions), fix implementation (2.0h — 0.5h work-stealing disabled via dask.config.set; 0.5h post-compute Spark validation assertion implemented with 0.1% tolerance; 0.5h 17-week historical backfill recalculated; 0.5h Redshift table updated with corrected 0.3%-reduced revenue figures), validation (1.0h — 0.5h post-fix verification run confirming zero work-stealing events and zero double-counted partitions; 0.5h all 17 corrected weekly aggregations validated against Spark and confirmed within 0.001% tolerance).

In a HourTab retainer log, the incident session reads: “Dask work-stealing groupby double-counting investigation: Dask 2024.3.1 on AWS EMR 7.1.0; 3× r5.2xlarge SSD + 1× r5.2xlarge EFS (worker-4); work-stealing enabled default; compute_weekly_revenue.py groupby(['product_category','region']).agg({'revenue':'sum','units_sold':'sum'}); EFS read latency 180–240ms vs. 20–30ms SSD (8–12×); tasks >1.5× median eligible for work-stealing; 60/200 partitions stolen per run; both original EFS and new SSD workers complete groupby accumulation; 60 partitions double-counted in distributed merge; 0.3% revenue inflation per run; 17 weeks (June 9–October 6 2026) undetected; no error raised; Dask dashboard all green; discovery: Spark one-off validation 0.3% discrepancy; fix: dask.config.set({'distributed.scheduler.work-stealing': False}) + post-compute Spark validation assertion (0.1% tolerance) + 17-week historical correction to Redshift; 4.5h.”

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: Dask developer retainers

See also: Apache Spark developer retainer · Apache Flink developer retainer · Apache Airflow developer retainer