Apache Airflow and SQLAlchemy: Connection Pool Exhaustion in Distributed Executor Environments
0 reputation · 17 Sept 2023, 09:41 UTC
0 reputation · 17 Sept 2023, 09:41 UTC
Apache Airflow relies on SQLAlchemy to manage the connection pool for its metadata database. In distributed environments, the interaction between the sql_alchemy_pool_size and sql_alchemy_max_overflow settings determines how the scheduler and workers handle concurrent database sessions.
When scaling worker nodes via the Celery or Kubernetes executors, there is a potential misalignment between the application-level pool limits and the database server's max_connections threshold. If workers maintain long-lived sessions or fail to return connections to the pool promptly during high-concurrency bursts, the system may encounter intermittent pool exhaustion.
Given these constraints, what is the recommended strategy for balancing pool size across distributed nodes to prevent QueuePool limit reached errors without exceeding the database server's hard connection limit? How does the session teardown behavior differ between the Celery and Kubernetes executors in this context?
33275 reputation · 17 Sept 2023, 17:26 UTC
First determine the hard connection limit of your metadata database (max_connections). Subtract a safety headroom (e.g., 10‑20 %) for Airflow’s own metadata sessions, monitoring tools, and any other services that share the same instance. The remaining connections are the budget for worker‑side SQLAlchemy pools.
Let W be the total number of concurrent worker processes (scheduler + executor workers). Divide the budget by W to obtain a per‑worker target T. Configure Airflow so that:
sql_alchemy_pool_size + sql_alchemy_max_overflow ≤ Tpool_size (e.g., 5) and allow max_overflow to absorb short bursts, but never exceed T.pool_pre_ping=True and a reasonable pool_recycle (e.g., 1800 s) to avoid stale connections in long‑lived workers.Scope the SQLAlchemy Engine and Session to the task execution context (e.g., inside the task function or a with Session(engine) as session: block). This ensures connections are returned to the pool promptly, regardless of executor type.
Worker processes are long‑lived and reuse the same Engine across many task invocations unless you recreate it per task. If a session is not explicitly closed or used in a context manager, the connection stays checked out for the lifetime of the worker (or until the worker is recycled). This can cause connections to accumulate during high‑concurrency bursts.
Each task runs in an isolated pod that is terminated after the task finishes. The Engine and any Session objects are created when the pod starts and are destroyed when the pod exits, automatically returning connections to the database. Even if a session is not closed, the pod’s termination releases the connection, making leaks less persistent than with Celery workers.
SELECT setting::int AS max_connections FROM pg_settings WHERE name = 'max_connections';
SELECT count(*) FROM pg_stat_activity WHERE state = 'active';
airflow.cfg or environment variables:echo $AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_SIZE echo $AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_OVERFLOW
(pool_size + max_overflow) * worker_count stays below the headroom‑adjusted max_connections.If you do not know your database’s max_connections, please provide that value; it directly influences the per‑worker pool target and may change the recommendation.
Use comments to ask for clarification. Post a solution as an answer.
33,275 reputation · 17 Sept 2023, 11:14 UTC
Each Airflow scheduler (including standby schedulers in HA) runs multiple parsing processes that each create a SQLAlchemy engine using the same sql_alchemy_pool_size and sql_alchemy_max_overflow settings. These processes consume connections alongside executor workers, so they must be added to the total concurrent worker count W when budgeting connections against the database's max_connections. Forgetting them can trigger QueuePool limit errors during heavy DAG parsing bursts, even if executor-side sizing appears correct.