Airflow - Beginners Tutorial

Airflow is a workflow engine from Airbnb. Airbnb developed it for its internal use and had recently open sourced it. In Airflow, the workflow is defined programmatically. Airflow document says that it's more maintainable to build workflows in this way, however I would leave it to the judgement of everyone. A developer, anyway would love anything programmatic. :) Airflow comes with a UI also and I can say that the UI is very clean and impressive. 

The main concept of airflow is a DAG (Directed Acyclic Graph). Yes, it's the same graph that you have seen in Maths, if you have seen it. A DAG contains vertices and directed edges. In a DAG, you can never reach to the same vertex, at which you have started, following the directed edges. Otherwise your workflow can get into an infinite loop. In workflow context, tasks can be defined as vertex and the sequence is represented with the directed edge. The sequence decides the order in which the tasks will be performed.

Make no mistake about the fact that airflow is just a workflow engine. It is only responsible for defining tasks and sequences. The details of task has to be handled by each task on its own. Airflow provides hooks for initiating tasks and has integration points to other systems. But at the end of the day, it's a workflow management system and no more than that. 

Oh!, one thing need to mention before we move ahead. Airflow is in Python and the workflows are also defined using Python. 

Installation

Airflow needs a home and we can give the home to any place. For this let's say we give it at home

export AIRFLOW_HOME=~/airflow

Install airflow

pip install airflow

Initialize database

airflow initdb

Start the webserver

airflow webserver -p 8080

Writing a DAG

Now let's write aworkflow in the form of a DAG. We will have four task t1, t2, t3 and t4. t4 will depend on t2 and t3. t2 and t3, in turn will depend on t1. Let's name the script helloWorld.py and put it in dags folder of airflow home. Create dags folder if it's not there.

from airflow import DAG

from airflow.operators import BashOperator

from datetime import datetime, timedelta

# Following are defaults which can be overridden later on

default_args = {

    'owner': 'lalit.bhatt',

    'depends_on_past': False,

    'start_date': datetime(2016, 4, 15),

    'email': ['lalit.bhatt@gmail.com'],

    'email_on_failure': False,

    'email_on_retry': False,

    'retries': 1,

    'retry_delay': timedelta(minutes=1),

}

dag = DAG('HelloWorld', default_args=default_args)

# t1, t2, t3 and t4 are examples of tasks created using operators

t1 = BashOperator(

    task_id='task_1',

    bash_command='echo "Hello World from Task 1"',

    dag=dag)

t2 = BashOperator(

    task_id='task_2',

    bash_command='echo "Hello World from Task 2"',

    dag=dag)

t3 = BashOperator(

    task_id='task_3',

    bash_command='echo "Hello World from Task 3"',

    dag=dag)

t4 = BashOperator(

    task_id='task_4',

    bash_command='echo "Hello World from Task 4"',

    dag=dag)

t2.set_upstream(t1)

t3.set_upstream(t1)

t4.set_upstream(t2)

t4.set_upstream(t3)

Testing and Installing DAG

Make sure script compiles. Go to dags folder and run

python helloWorld.py

Some useful command lines

List all DAGs'

airflow list_dags   

List Tasks for the given DAG

airflow list_tasks HelloWorld

Backfilling

airflow backfill HelloWorld -s 2015-04-12 -e 2015-04-15

Individual tasks can be tested also

airflow test HelloWorld task_1 2016-04-15

For details see the following video

Reference: https://github.com/airbnb/airflow . 

Also see the discussion at https://github.com/airbnb/airflow/issues/289  . This gives an idea about the possible future of airflow