Apache Airflow with 3 Celery workers in docker-compose

UPD from 20 July 2020: this article is pretty old and when I write it there was not exist the official Docker image so in article I used puckel/docker-airflow. But now, Apache Airflow already created and supported production-ready official docker image. You can read more about it here: https://github.com/apache/airflow/blob/master/IMAGES.rst#airflow-docker-images.

I prepared docker-compose on official Docker image you can find it here: https://github.com/xnuinside/airflow_in_docker_compose/blob/master/docker-compose-with-celery-executor.yml (with Apache Airflow version 1.10.11) and .env file for it https://github.com/xnuinside/airflow_in_docker_compose/blob/master/.env . I hope I will have time soon to create article with explanation.

Previous, we describe how to run simple Apache Airflow infrastructure with PostgreSQL DB, Airflow Webserver and Scheduler. Next logical step is to cover the most popular case — when we use CeleryExecutor as a solution for production.

Apache Airflow with three Celery Workers and docker-compose

If you don’t know that is Celery: http://www.celeryproject.org/ (but pay attention, that Celery used only as workers in Airflow, not as Scheduler)

Usually, you don’t want to use in production one Celery worker — you have a bunch of them, for example — 3. And this causes some cases, that do not exist in the work process with 1 worker. Let me show an example.

Imagine, that you have a workflow, where you have such a piece of logic:

1. Load data from some API
2. parse it and format
3. write results to DB.

In case when you have only one worker, you always had a guarantee, that data that you load will be on the same machine where Airflow will start next task (because of you have only one machine at all). But what will happen if you have 3 workers? The first task will run on one machine and load data on worker1, the second task will start on worker 2, where you will not have this data.

Of course, this case has different workarounds: you can have shared between workers NFS disk, you can have a SubDAG for this part of code with SequentialExecutor to guarantee that all task was executed in the same machine.

But this example needs just to show that you will have different cases with Airflow if you work with one worker and 3 workers. So, very important to develop local and test infrastructure that map 1-to-1 in real production or you can have interesting surprises after release.

So, let’s change our infrastructure to work with CeleryExecutor. First of all, we need to modify in airflow.cfg [core] section variable executor to:

executor = CeleryExecutor

To work with Celery we also need to set up Celery Backend (where we will save a tasks’ results) and Broker (queue for our tasks). This is need to be defined in airflow.cfg.

In section ‘[celery]’ define variable ‘result_backend’ and ‘broker_url’ (search them in airflow.cfg file):

For backend, we will use the same DB as for Airflow backend

In our example:

result_backend = db+postgresql://airflow:airflow@postgres:5432/airflow

And we will use Redis for Broker.

In our example it will be:

broker_url = redis://redis:6379/0

For Redis, we will use an official image — https://hub.docker.com/_/redis.

Let’s add to our Redis in docker-compose.yml (in services: section, after declarations with db, webserver and scheduler, that we already have). Pay attention, that we use as base docker-compose.yml, that was created in the first article — Quick guide: How to run Apache Airflow cluster in docker-compose

redis:
  image: redis:5.0.5

And to add 3 workers services. I will be not original, I named services in docker-compose as worker_1, worker_2, worker_3 and I will use the same names for CELERY_HOSTNAME.

Somebody can ask a question, why we do this and add different names for workers, and not just use docker-compose scale + ngnix.

Because ‘docker-compose scale + ngnix’ we will use when we need to have fault tolerance or we solve performance issues, but this is an additional ‘tool’.

When we have several different named workers we can use queues to split our DAGs between workers by priorities or DAG’s functionality.

For example, one worker is placed on VM, that also has installed JVM and resources for it, for example, to run Spark — we want always execute on this worker only DAGs with Spark. So you have only one VM with JVM and it’s enough for you, other workers run Ruby processes or only pure python. But, if you need fault tolerance you can scale your first ‘java-airflow worker’ with ‘compose scale + ngnix’ or some another way to have 3 instances of this airflow worker with JVM, but you will anyway have another ‘ruby-airflow worker’ or ‘pure-python-airflow worker’, that can also be scaled up with nginx. So, this is just scale — level for different purposes.

But this article not about queues and split DAGs between workers. Let’s return to our docker-compose.yml

For workers, you need to add this to your docker-compose file services section :

worker_1:
  build: .
  restart: always
  depends_on:
    - postgres
  volumes:
    - ./airflow_files/dags:/usr/local/airflow/dags
  entrypoint: airflow worker -cn worker_1
  healthcheck:
    test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-worker.pid ]"]
    interval: 30s
    timeout: 30s
    retries: 3
worker_2:
  build: .
  restart: always
  depends_on:
    - postgres
  volumes:
    - ./airflow_files/dags:/usr/local/airflow/dags
  entrypoint: airflow worker -cn worker_2
  healthcheck:
    test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-worker.pid ]"]
    interval: 30s
    timeout: 30s
    retries: 3
worker_3:
  build: .
  restart: always
  depends_on:
    - postgres
  volumes:
    - ./airflow/dags:/usr/local/airflow/dags
  entrypoint: airflow worker -cn worker_3
  healthcheck:
    test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-worker.pid ]"]
    interval: 30s
    timeout: 30s
    retries: 3

Great, now we have the last step in modification of our docker-compose file before the run — we need to add Celery Flower. To get possible understand that's going on with Celery workers — you need to have Flower. The Flower is a Celery monitoring tool. If you don’t know that it is, check doc: https://flower.readthedocs.io/en/latest/.

Flower used with UI, so it’s important don’t forget to share a port with ‘ports’ in docker-compose file. After all workers declarations in the ‘services’ section, add:

flower:
  build: .
  restart: always
  depends_on:
    - postgres
  volumes:
    - ./airflow_files/dags:/usr/local/airflow/dags
  entrypoint: airflow flower
  healthcheck:
    test: ["CMD-SHELL", "[ -f /usr/local/airflow/airflow-flower.pid ]"]
    interval: 30s
    timeout: 30s
    retries: 3
  ports:
    - "5555:5555"

Now, we are ready to rebuild and up to our cluster:

docker-compose down  # if your cluster is active
docker-compose up --build

You will see in console something like this for each worker “worker_name ready”:

Successful console traceback with Celery Worker

Now, check Flower, we need to see our workers in Flower UI. To access Flower UI use http://localhost:5555. You will see a list of workers:

Empty Celery Flower UI with 3 workers

Now, time to check the correct task execution. Enter the Airflow UI http://localhost:8080 and run you test DAG (we use BashOperator example DAG in the previous article).

If all goes right, you will see the active and processed task in Flower UI, like this:

Flower UI after Airflow tasks execution

And of course, in main Airflow UI you will see tasks changed statuses.

And that is it. I hope the tutorial was useful. Leave your comments if you will glad to have some additional info or separate article about some topic.

All sources you can find here (all files special for usage with CeleryExecutor marked with ‘celery_executor’): https://github.com/xnuinside/airflow_in_docker_compose

Next:

Install Python dependencies to docker-compose cluster without re-build images