Apache Airflow Master and Worker Nodes

Setting up an Apache Airflow Master instance using docker.

Primary Airflow application and dependencies (pg, redis, etc) docker-compose.yaml
Primary node hosts the web interface and flower (celery node monitoring)
Primary node also contains an embedded Airflow Celery Worker node
Remote Celery Worker nodes to scale out the workers docker-compose-worker.yaml

Creating the Airflow Master Node

Copy the below .env and docker-compose.yaml and update them as required.
Run docker-compose up -d , it will take 2-3 mins for PG, Redis, etc to init and come online
Access Airflow Web at http://$hostname:8080 using username/pass: airflow/airflow
Access Flower at http://$hostname:5555, no credentials are required
You can run jobs immediately by as a local celery worker is automatically spawned on the same host

Creating Airflow Worker Nodes

Copy the below files .env (edit HOSTNAME) and docker-compose-worker.yaml
Run docker-compose -f docker-compose-worker.yaml up to initialize the worker.
Check flower UI to ensure the worker is detected correctly
Bugs: Ensure ./logs folder is writeable by the airflow celery user by running chown $AIRFLOW_UID ./logs. This is required only on the worker nodes, the master node permissions are set correctly.

Common `.env`

.env file holding common variables

# Filename: .env

COMPOSE_PROJECT_NAME=airflow
# Use arunsupe as the airflow user so file perms in /data don't get messed up
AIRFLOW_UID=1000

# Generate using "openssl rand -hex 20"
FERNET_KEY=ca85686f85fa83415137d6c6486802bb0e472ded
SECRET_KEY=ca85686f85fa83415137d6c6486802bb0e472ded

# PG and Redis are on bile, these are tailscale IPs
X_PG_HOST='100.68.204.9'
X_REDIS_HOST='100.68.204.9'

# This needs to be reset to the local hostname on every node
HOSTNAME=bile

Main Airflow with All Dependencies

docker-compose.yaml for the main Airflow application and dependencies. This also auto-starts a Celery Worker node in the same host.

# Filename: docker-compose.yaml
# References:
# https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html
# https://stackoverflow.com/questions/68194327/how-to-configure-celery-worker-on-distributed-airflow-architecture-using-docker/68198920#68198920
---
version: '3'
x-airflow-common:
  &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.4.2}
  environment:
    &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: CeleryExecutor
    AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@${X_PG_HOST}/airflow
    # For backward compatibility, with Airflow <2.3
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@${X_PG_HOST}:5432/airflow # 5432 is default postgres port
    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@${X_PG_HOST}:5432/airflow
    AIRFLOW__CELERY__BROKER_URL: redis://:@${X_REDIS_HOST}:6379/0
    AIRFLOW__CELERY__WORKER_CONCURRENCY: 16
    AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY}
    AIRFLOW__WEBSERVER__SECRET_KEY: ${SECRET_KEY}
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
    AIRFLOW__API__AUTH_BACKENDS: "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
    _PIP_ADDITIONAL_REQUIREMENTS: ${_PIP_ADDITIONAL_REQUIREMENTS:-}
    AIRFLOW__CORE__PARALLELISM: 64
    AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG: 32
    AIRFLOW__SCHEDULER__PARSING_PROCESSES: 4
  # Add all hosts which will be Celery workers 
  extra_hosts:
    - "bile:100.68.204.9"
    - "celia:100.88.68.37"
    - "sully:100.71.129.65"
    - "mike:100.95.196.87"
  volumes:
    - ./dags:/opt/airflow/dags
    - ./logs:/opt/airflow/logs
    - ./plugins:/opt/airflow/plugins
#   - ./docker:/usr/bin/docker:ro
    - /data:/data
    - /etc/localtime:/etc/localtime:ro
  user: "${AIRFLOW_UID:-50000}:0"
  depends_on:
    &airflow-common-depends-on
    redis:
      condition: service_healthy
    postgres:
      condition: service_healthy

services:
  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    ports:
      - 5432:5432
    volumes:
      - ./postgres-db-volume:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 5s
      retries: 5
    restart: always

  redis:
    image: redis:latest
    ports:
      - 6379:6379
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 30s
      retries: 50
    restart: always

  airflow-webserver:
    <<: *airflow-common
    command: webserver
    ports:
      - 8080:8080
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:8080/health"]
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-scheduler:
    <<: *airflow-common
    command: scheduler
    healthcheck:
      test: ["CMD-SHELL", 'airflow jobs check --job-type SchedulerJob --hostname "${HOSTNAME}"']
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-worker:
    <<: *airflow-common
    command: celery worker
    hostname: ${HOSTNAME}
    ports:
      - 8793:8793
    healthcheck:
      test:
        - "CMD-SHELL"
        - 'celery --app airflow.executors.celery_executor.app inspect ping -d "celery@${HOSTNAME}"'
      interval: 10s
      timeout: 10s
      retries: 5
    environment:
      <<: *airflow-common-env
      # Required to handle warm shutdown of the celery workers properly
      # See https://airflow.apache.org/docs/docker-stack/entrypoint.html#signal-propagation
      DUMB_INIT_SETSID: "0"
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-triggerer:
    <<: *airflow-common
    command: triggerer
    healthcheck:
      test: ["CMD-SHELL", 'airflow jobs check --job-type TriggererJob --hostname "${HOSTNAME}"']
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

  airflow-init:
    <<: *airflow-common
    entrypoint: /bin/bash
    # yamllint disable rule:line-length
    command:
      - -c
      - |
        function ver() {
          printf "%04d%04d%04d%04d" ${1//./ }
        }
        airflow_version=$(AIRFLOW__LOGGING__LOGGING_LEVEL=INFO && gosu airflow airflow version)
        airflow_version_comparable=$(ver ${airflow_version})
        min_airflow_version=2.2.0
        min_airflow_version_comparable=$(ver ${min_airflow_version})
        if (( airflow_version_comparable < min_airflow_version_comparable )); then
          echo
          echo -e "\033[1;31mERROR!!!: Too old Airflow version ${airflow_version}!\e[0m"
          echo "The minimum Airflow version supported: ${min_airflow_version}. Only use this or higher!"
          echo
          exit 1
        fi
        if [[ -z "${AIRFLOW_UID}" ]]; then
          echo
          echo -e "\033[1;33mWARNING!!!: AIRFLOW_UID not set!\e[0m"
          echo "If you are on Linux, you SHOULD follow the instructions below to set "
          echo "AIRFLOW_UID environment variable, otherwise files will be owned by root."
          echo "For other operating systems you can get rid of the warning with manually created .env file:"
          echo "    See: https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#setting-the-right-airflow-user"
          echo
        fi
        one_meg=1048576
        mem_available=$(($(getconf _PHYS_PAGES) * $(getconf PAGE_SIZE) / one_meg))
        cpus_available=$(grep -cE 'cpu[0-9]+' /proc/stat)
        disk_available=$(df / | tail -1 | awk '{print $4}')
        warning_resources="false"
        if (( mem_available < 4000 )) ; then
          echo
          echo -e "\033[1;33mWARNING!!!: Not enough memory available for Docker.\e[0m"
          echo "At least 4GB of memory required. You have $(numfmt --to iec $((mem_available * one_meg)))"
          echo
          warning_resources="true"
        fi
        if (( cpus_available < 2 )); then
          echo
          echo -e "\033[1;33mWARNING!!!: Not enough CPUS available for Docker.\e[0m"
          echo "At least 2 CPUs recommended. You have ${cpus_available}"
          echo
          warning_resources="true"
        fi
        if (( disk_available < one_meg * 10 )); then
          echo
          echo -e "\033[1;33mWARNING!!!: Not enough Disk space available for Docker.\e[0m"
          echo "At least 10 GBs recommended. You have $(numfmt --to iec $((disk_available * 1024 )))"
          echo
          warning_resources="true"
        fi
        if [[ ${warning_resources} == "true" ]]; then
          echo
          echo -e "\033[1;33mWARNING!!!: You have not enough resources to run Airflow (see above)!\e[0m"
          echo "Please follow the instructions to increase amount of resources available:"
          echo "   https://airflow.apache.org/docs/apache-airflow/stable/howto/docker-compose/index.html#before-you-begin"
          echo
        fi
        mkdir -p /sources/logs /sources/dags /sources/plugins
        chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags,plugins}
        exec /entrypoint airflow version
    # yamllint enable rule:line-length
    environment:
      <<: *airflow-common-env
      _AIRFLOW_DB_UPGRADE: 'true'
      _AIRFLOW_WWW_USER_CREATE: 'true'
      _AIRFLOW_WWW_USER_USERNAME: ${_AIRFLOW_WWW_USER_USERNAME:-airflow}
      _AIRFLOW_WWW_USER_PASSWORD: ${_AIRFLOW_WWW_USER_PASSWORD:-airflow}
      _PIP_ADDITIONAL_REQUIREMENTS: ''
    user: "0:0"
    volumes:
      - .:/sources

  airflow-cli:
    <<: *airflow-common
    profiles:
      - debug
    environment:
      <<: *airflow-common-env
      CONNECTION_CHECK_MAX_COUNT: "0"
    # Workaround for entrypoint issue. See: https://github.com/apache/airflow/issues/16252
    command:
      - bash
      - -c
      - airflow

  # Flower - Celery monitoring tool
  flower:
    <<: *airflow-common
    command: celery flower
    ports:
      - 5555:5555
    healthcheck:
      test: ["CMD", "curl", "--fail", "http://localhost:5555/"]
      interval: 10s
      timeout: 10s
      retries: 5
    restart: always
    depends_on:
      <<: *airflow-common-depends-on
      airflow-init:
        condition: service_completed_successfully

Airflow Celery Worker

docker-compose-worker.yaml for an Airflow Worker node. This sets up a celery worker node which attaches to the main Airflow application above.