220 Airflow DAGs, Duplicate Rows, and a 02:00 Out-of-Memory

The duplicates are the retries doing their job on tasks that were never idempotent. A Dataproc or Dataflow job that appends to BigQuery, re-run after a connection error, appends the same day's rows a second time. The connection errors themselves are usually resource exhaustion - 80 concurrent tasks hammering the Airflow metadata database and the Dataproc/Dataflow APIs - so the retries are a symptom of the fan-out problem, not an independent bug.

Make the unit of write the partition, and make partition writes atomic. Each task computes into a run-scoped staging location keyed by the logical date (execution_date, not now()), then replaces exactly that date's partition: WRITE_TRUNCATE for a full-day rewrite, or a MERGE on the natural key where the day is incrementally updated. A retry then overwrites rather than appends, and the result is identical whether the task ran once or three times.

Guard the boundaries: max_active_runs=1 per DAG so two scheduled runs cannot overlap on the same partition, and a check at task start that the target partition is not already marked complete with a matching run_id - skip rather than redo when it is.

Keep retries on. Turning them off hides the flakiness; the correct response to a flaky-but-idempotent task is to retry it, then track the retry rate as a metric so the underlying cause stays visible. Six DAGs a night is 2-3% and should be trending down, not normalized.

Composer worker memory is consumed by task processes, and 80 parallel tasks - especially Python tasks that hold a client and parse responses - exhaust 3 workers. When the workers thrash, the scheduler (which shares the environment) degrades, which is also why other DAGs start throwing connection errors in the same window. One fan-out DAG is taking down the platform.

Fixes, in order of leverage:

1. Move the fan-out out of Airflow. Airflow is an orchestrator, not a compute engine. If the 80 tasks each submit a small Dataproc/Dataflow job, submit one job that does the 80 units of work internally, or use a single Beam pipeline with a dynamic work source. Airflow then tracks one task instead of 80.
2. Bound concurrency with a pool. Put the fan-out behind a named pool sized to what the environment can actually sustain - 15 or 20, not 80 - so the queue absorbs the burst instead of the workers.
3. Use deferrable operators for anything that waits on a long-running job, so the worker slot is released while the job runs rather than being held for 40 minutes.
4. Split the DAG into a small coordinator plus per-domain DAGs triggered by dataset/asset dependencies, so a failure is contained to one domain.

Measure it: record worker memory and task concurrency as dashboards before and after. If the environment is still tight after the fan-out is fixed, the residual is a sizing question - but resize last, because resizing hides the design problem.

Alerting: assert on data, not on task states. Airflow's own failure callbacks are unreliable when the scheduler is the thing under stress, and a task can be 'successful' while producing an empty table. Instead publish a contract and check it independently.

• Each pipeline writes a completion marker - a row in a pipeline_runs table with dataset, partition_date, run_id, row_count, status - as its final action.
• A Cloud Monitoring log-based alert or a scheduled query fires if, at 05:30, the expected partition_date marker is missing or row_count is below the trailing baseline by more than a set percentage.
• Route task-level failures from Cloud Logging to Pub/Sub to the on-call channel as a secondary signal, so you get two independent paths.

Why 05:30 and not 09:00: the deadline is data availability, so alert on the freshness SLO with a buffer before the consumer needs it. State the SLO as 'partition complete by 05:30, 99% of nights' and alert on burn rate, not on single failures - otherwise the 6 known-flaky DAGs generate noise that trains people to ignore the channel.

Idle workers: the Composer worker count is a static environment setting; it does not shrink between runs, so a nightly workload pays for 24 hours. Options, in order: reduce the environment to the smallest size that survives the peak (which is much smaller once the 80-task fan-out is gone), move long-running compute to deferrable operators so slots are not held, and if the daytime workload is near zero, evaluate whether the scheduling belongs on a lighter orchestrator. Do not commit to a 3-year commitment on the current environment size before the fan-out fix lands - you would be committing to the overprovisioning.