AAtsushi's Blog
Airflow

I Didn't Understand Airflow, So I Studied on Udemy

→ 日本語版を読む

Overview

The Airflow Tutorial didn't seem like something I could understand quickly, so I tried a Udemy course. It was an English course, which was a bit tough, so I only got through about halfway. I got a rough understanding of the very basics.

https://www.udemy.com/course/the-complete-hands-on-course-to-master-apache-airflow

So here are some rough notes. Apologies if any understanding is off.

Code will be in a separate post.

What's Great About Airflow

  • Everything can be written in Python (no XML)
  • Scalable
  • Good UI
  • Extensible via plugins

Components

There are four core components:

  • Web server: The Airflow UI. Built with Flask.
  • Scheduler: Passes tasks to Workers.
  • Metastore: Any DB compatible with SQL Alchemy works. For example, Postgres, MySQL, Oracle DB, SQL Server, etc. The metadata store holds metadata about data pipelines, tasks, Airflow users, etc.
  • Triggerer: Executes specific tasks. Details later.

Other components:

  • Executor: K8s clusters require a K8s Executor, Celery requires a Celery Executor.
  • Queue: A queue for holding tasks.
  • Worker: Executes tasks.

(*1): Celery is apparently a distributed processing Python framework.

Concepts in Airflow

  • Workflow: A workflow is the concept encompassing DAGs, operators, and tasks. A workflow consists of one or more DAGs (probably).
  • Task / Task instance: A Task is the name for executing an Operator. A Task instance... I'm not entirely sure, but it seems to be instantiated when a Task is processed.
  • Operator: There are three types of operators:
    • Action operator: e.g., BashOperator
    • Transfer operator: Transfers data. For example, transferring data from MySQL to Redshift.
    • Sensor operator: Waits for processing to be ready. For example, FileSensor waits for a file.
  • DAG: Defines the order between tasks.

Architectures

Single-node architecture

  • The simplest architecture for running Airflow on one machine.
  • Tasks are processed by machine processes.

Multi-node architecture

  • Used in production. Has redundancy and can handle large workloads.
  • Uses Celery or K8s for multi-node processing.
  • Node 1 has the Web server, Scheduler, and Executor; Node 2 has the Metastore and Queue. Additionally, there are Worker nodes.
  • The Executor puts tasks in the Queue, and each Worker node pulls tasks from the Queue.
  • With Celery Executor, the Queue uses Rabbit MQ or Redis.
  • Celery Worker nodes can be started with the airflow celery worker command.

How Airflow Works

  • When the Scheduler triggers a DAG, a DAG Run object with Running status is created.
  • When the DAG Run object executes the first Task, that Task becomes a Task Instance object.
  • The Task Instance object starts with None status but quickly becomes Scheduled.
  • Then the Scheduler sends the Task Instance object to the Executor's Queue.
  • The Task Instance object added to the Queue gets Queued status.
  • Then the Executor creates a subprocess to execute the task.
  • The Task Instance object then has Running status.
  • When the task completes, the Task Instance object's status becomes Success or Failed.
  • Then the Scheduler checks if there are tasks to execute, and if not — meaning the DAG is complete — the DAG Run object becomes Success.
  • The statuses of DAG Run objects and Task Instance objects can be checked in the Web UI.

DAG Run status transitions:

Running → Success / Failed

Task Instance object status transitions:

None → Scheduled → Queued → Running → Success / Failed

Where to Place Airflow Code

  • Create Airflow code as a .py file and place it in the dags folder.
  • However, placing it in ~/airflow/dags doesn't immediately reflect in the Airflow UI.
  • The Scheduler checks the DAG folder every 5 minutes, parses any .py files found, and reflects them in the Airflow UI.
  • The Scheduler also parses file modifications every 30 seconds, so it can take up to 30 seconds to appear in the Airflow UI.

Airflow UI

Explanation of what can be checked in each view.

DAGs view

  • Toggle: Start/stop a DAG.
  • DAG name
  • Tags
  • Owner
  • Runs status: Status of DAG Run objects.
  • Schedule: Cron-style startup time setting (note: different meaning from Cron).
  • Last Run: The last time the DAG was executed.
  • Next Run: The next time the DAG will be executed.
  • Recent Tasks: Status of executed tasks.
  • Actions: Start DAG, delete DAG metadata.
  • Links

Landing view

  • Plots the time taken to execute a DAG on a daily basis.
  • Allows evaluating how much the DAG execution time has decreased (or increased).
  • For example, you can check how much time was reduced after increasing worker instances.

Gantt view

  • Displays how much time each task took in Gantt chart format.
  • When tasks overlap in time, it indicates they are running in parallel.
  • If the Gantt chart looks like sequential processing, it needs to be modified to run in parallel.

Code view

  • Displays the DAG's code.
  • Verify whether modified code has been reflected.

Graph view

  • Shows dependencies between tasks.
  • The color of the task box indicates the task's execution result.

DAG (Directed Acyclic Graph) Terminology

  • Node: Represents a task.
  • Edge: Represents the dependency between tasks.

Notes on Defining Operators

Map one process to one Operator.

For example, if you try to process both data cleaning and data transformation in the same single Python operator, and data transformation fails and you want to retry, the operator also includes data cleaning, so retrying will re-execute data cleaning (which already succeeded) along with data transformation.

About Providers

Airflow has a modular architecture.

When Airflow is installed, only the main operators like Python operator and Bash operator are installed.

For example, to use AWS operators, you need to install the Amazon provider.

Install the providers you want to use separately.

For what providers are available, refer to:

https://registry.astronomer.io/

About Connections

Connection information for DBs etc. can be registered in the Airflow UI.

Select Connections from the Admin tab in the Airflow UI and register new connection information.

About Sensors

Important Sensor parameters:

poke_interval

  • By default, the Sensor checks whether the condition is satisfied every 60 seconds.

timeout

  • Set to 7 days by default. If the condition is not met after 7 days of waiting, it becomes Failed.

Sensor examples:

HTTP Sensor: Checks whether an API is running.

About Hooks

Abstracts connections to external systems for easy connectivity.

If an Operator can't connect, it's worth checking if a Hook can.

Specifying Task Dependencies

Use >> or <<.

Like extract_website >> process_website >> store_user.

    extract_website = SimpleHttpOperator(
        ...
    )

    process_website = PythonOperator(
        ...
    )

    store_user = PythonOperator(
        ...
    )

    extract_website >> process_website >> store_user

Or use set_upstream and set_downstream.

To branch a DAG, use Branch Operator. Refer to here.

DAG Scheduling

To run a DAG every 2 minutes, write */2 * * * *:

dag = DAG("tutorial", ..., schedule_interval="*/2 * * * *")

Backfilling Mechanism

Using the airflow dags backfill command runs the DAG for periods before the start_date.

Catchup Mechanism

Setting catchup=True will retroactively execute any DAG Runs that haven't been run since the start_date.

with DAG('tutorial',..., catchup=True) as dag:

Config

To change the Executor and other settings, edit the airflow.cfg config file.

Config file content can be overridden with environment variables. For example, the AIRFLOW__CORE__EXECUTOR environment variable can override the Executor setting.

Executor Types

Sequential Executor

Can only process one task at a time.

LocalExecutor

If using SQLite, only Sequential Executor can be used.

By using PostgreSQL or MySQL instead of SQLite, LocalExecutor becomes available.

Like Sequential Executor, LocalExecutor has only one Worker, but it can process multiple tasks in parallel through multiprocessing.

In other words, LocalExecutor can handle more tasks in parallel through scale-up (not scale-out).

When using LocalExecutor, configure as follows:

executor=LocalExecutor
sql_alchemy_conn=postgresql+psycopg2://<user>:<password>@<host>/<db>

CeleryExecutor

Uses Celery to process tasks across multiple Workers.

With CeleryExecutor, a CeleryQueue is a component in the architecture.

This Celery Queue has two types of DBs: Broker and ResultBackend.

The Broker receives tasks from the Scheduler and assigns them to Workers.

When processing is complete on a Worker, the result is saved to the ResultBackend.

Configuration when using CeleryExecutor:

executor=CeleryExecutor
sql_alchemy_conn=postgresql+psycopg2://<user>:<password>@<host>/<db>
celery_result_backend=postgresql+psycopg2://<user>:<password>@<host>/<db>
celery_broker_url=redis://@redis:6379/0

Trivia

Setting load_examples = False in airflow.cfg hides the example DAGs from the Airflow UI.

Sub DAG

[IMPORTANT] SubDAG is complex and was deprecated in Airflow 2.2. Use Task Group instead of SubDAG.

SubDAG allows grouping multiple tasks into a single sub-DAG.

Grouping tasks into one simplifies operations.

To use Subdag, the following are required:

  • Create a subdags folder under the dags folder.
  • Create a .py file in the subdags folder and define a function there.
  • The function takes three arguments: parent_dag_id, child_dag_id, args.
  • Define a DAG within the function.
with DAG (f"{parent_dag_id}.{child_dag_id}", ...   ) as dag:
...
return dag
  • Call the SubDAG using SubDAGOperator in the main .py file.

Task Group

  • Create a groups folder under the dags folder.
  • Create a .py file in the groups folder and define a function there.
  • No function arguments are needed.
  • Define a TaskGroup within the function.
with TaskGroup ("task_group_id", tooltip="hogehoge") as group:
...
return group
  • Call the defined function in the main .py file.
← Back to articles