Dynamic Task Mapping With Uneven Batches
A DAG uses @task with .expand() to process a list of customer IDs fetched at runtime. The list varies from 50 to 5,000 IDs. Each mapped task calls a rate-limited API that allows 10 requests per second per client. The DAG has max_active_tasks=16 at the DAG level and runs hourly. The team observes HTTP 429 errors during large batches and long tail latency during small batches because the API client is initialized per task. They also want to skip IDs that were already processed in the previous hour without failing the DAG.
1. Explain how mapped task concurrency is controlled and why the current settings produce 429s on large batches.
Read the worked answer
Dynamic task mapping creates one task instance per expanded input at runtime. Concurrency is bounded by max_active_tasks on the DAG and by pool slots. With max_active_tasks=16 and a rate limit of 10 requests per second per client, 16 concurrent tasks each calling the API can exceed the limit if each task makes more than one request per second. The limit is per client, so if each mapped task constructs its own client, the effective limit is multiplied by the number of concurrent tasks.
The 429s appear on large batches because more mapped instances compete for the same 16 slots and the same API quota. On small batches, the overhead of client initialization dominates, producing long tail latency.
The fix is to assign mapped tasks to a dedicated pool sized to the API quota, use exponential backoff with jitter on 429 responses, and share or reuse the client where possible. Consider batching IDs into chunks so each mapped task handles several IDs, reducing per-task overhead.
2. Design the skip logic for already-processed IDs and the retry behavior for rate limits.
Read the worked answer
Fetch the previous hour's processed IDs at the start of the run and compute the difference before expanding. Use a @task.short_circuit or a mapped task that raises AirflowSkipException for IDs already processed, so the DAG records them as skipped rather than failed. Keep the processed-ID set in durable storage keyed by logical date so retries are deterministic.
For rate limits, set retries with retry_exponential_backoff=True and a max_retry_delay, and catch 429 specifically to raise a retryable exception. Do not retry non-idempotent writes blindly; make the API call idempotent with a request key. Set execution_timeout so stuck calls do not hold slots.
Finally, add a pool with slots equal to the API's allowed concurrency and assert in tests that mapped task count times requests per second stays under the quota.