Redis – How to configure celery worker on distributed airflow architecture using docker-compose?

Far from being the "ultimate set-up", these are some settings that worked for me using the docker-compose from Airflow in the core node and the workers:

Main node:

• The worker nodes have to be reachable from the main node where the Webserver runs. I found this diagram of the CeleryExecutor architecture to be very helpful to sort things out.

  When trying to read the logs, if they are not found locally, it will try to retrieve them from the remote worker. Thus your main node may not know the hostname of your workers, so you either change how the hostnames are being resolved (hostname_callable setting, which defaults to socket.getfqdn ) or you just simply add name resolution capability to the Webserver. This could be done by adding the extra_hosts config key in the x-airflow-common definition:

---
version: "3"
x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
    ...# env vars
  extra_hosts:
    - "worker-01-hostname:worker-01-ip-address" # "worker-01-hostname:192.168.0.11"
    - "worker-02-hostname:worker-02-ip-address"

*Note that in your specific case where you have a shared drive, so I think the logs will be found locally.

• Define parallelism, DAG concurrency, and scheduler parsing processes. Could be done by using env vars:

x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
    AIRFLOW__CORE__PARALLELISM: 64
    AIRFLOW__CORE__DAG_CONCURRENCY: 32
    AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4

Of course, the values to be set depend on your specific case and available resources. This article has a good overview of the subject. DAG settings could also be overridden at DAG definition.

Worker nodes:

• Define worker CELERY__WORKER_CONCURRENCY, default could be the numbers of CPUs available on the machine (docs).

• Define how to reach the services running in the main node. Set an IP or hostname and watch out for matching exposed ports in the main node:

x-airflow-common: &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment: &airflow-common-env
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CELERY__WORKER_CONCURRENCY: 8
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@main_node_ip_or_hostname:5432/airflow # 5432 is default postgres port
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@main_node_ip_or_hostname:5432/airflow
  AIRFLOW__CELERY__BROKER_URL: redis://:@main_node_ip_or_hostname:6379/0

• Share the same Fernet Key and Secret Key reading them from an ".env" file:

  environment: &airflow-common-env
    AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY}
    AIRFLOW__WEBSERVER__SECRET_KEY: ${SECRET_KEY}

  env_file:
    - .env

.env file: FERNET_KEY=jvYUaxxxxxxxxxxxxx=

• It’s critical that every node in the cluster (main and workers) has the same settings applied.

• Define a hostname to the worker service to avoid autogenerated matching the container id.

• Expose port 8793, which is the default port used to fetch the logs from the worker (docs):

services:
  airflow-worker:
    <<: *airflow-common
    hostname: ${HOSTNAME}
    ports:
      - 8793:8793
    command: celery worker
    restart: always

• Make sure every worker node host is running with the same time configuration, a few minutes difference could cause serious execution errors which may not be so easy to find. Consider enabling NTP service on host OS.

If you have heavy workloads and high concurrency in general, you may need to tune Postgres settings such as max_connections and shared_buffers. The same applies to the host OS network settings such as ip_local_port_range or somaxconn.

In any issues I encountered during the initial cluster setup, Flower and the worker execution logs always provided helpful details and error messages, both the task-level logs and the Docker-Compose service log i.e: docker-compose logs --tail=10000 airflow-worker > worker_logs.log.

Hope that works for you!