Data Engineering Series #3: Apache Airflow - the modern Workflow management tool. Getting Started
Tags: opensource, python, beginners, career
Published: 2021-03-30 04:00:13


###  **Why such attention towards Airflow ?**

###### Interest rate of airflow. Source: Google Trends
![Source: Google Trends](https://dev-to-uploads.s3.amazonaws.com/i/y3v4zut344lht5r78tkb.png)

<i>"The software industry has seen a notable rise in the development of tools to manage data. These tools range from storage solutions that house data to pipelines that transport data".</i><br>

Data-driven companies like Airbnb, Quizlet rely on these data pipelines to resolve some tedious tasks like 

- Scheduling
- Testing
- Handling errors
- Versioning
- Scaling

the current data flow set up.

We call these pipelines data workflows. And a well-known tool in this space is <i>Apache Airflow</i>.

Airflow solves those tedious tasks by,
###### click on a feature if you wish to jump to its section directly.

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/o4fxoqkq0pintsjx0wji.png)](#dag)

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/o8slaav00tgwbt4u9xy3.png)](#retry)

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ei4oh9sued6ckuzc31yv.png)](#alert)

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/2ejrss7rl0dpeih9ymj0.png)](#web)

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ai9jl8836y0tzuash0ro.png)](#sch)

[![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/lucmwct3d8javiity63g.png)](#api)

And the best part of all, Airflow is an Open Source tool and has a rapidly growing user base and contributions 🙂


------------

Without further ado,
### **Let's deep dive into the workflow framework**

And create a minimalistic workflow pipeline that leverages all the Airflow features I had listed above. 

<a name = "dag"></a><h4>✨ DAGS</h4>
Every workflow is constructed as a Directed Acyclic Graph (DAG). 
DAG is created using a .py file which should ideally have three sections configured
1. DAG configuration - `default_args`
2. DAG Instance `DAG()`
3. DAG Tasks - `Operators`

<a name = "def"></a><br>
#### DAG Configuration - `default_args`
`default_args` is used to set properties <i>(arguments)</i> that are common for all the tasks.
<a name = "defcd"></a>
```py
# Initializing the default arguments that we'll pass to our DAG
default_args = {
    'owner': 'Srinidhi',
    'start_date': datetime(2018, 1, 1),
    'retries': 1, 
    'on_failure_callback': task_failure_alert
}
```
<a name = "dagcd"></a>
#### DAG Instance - `DAG()`
```py
# Creating a DAG instance
my_workflow_dag = DAG(
    'first_dag',
    default_args=default_args,
    description='Sample DAG for DE Series',
    schedule_interval='0 0 * * *',
)
```

<a name = "opr"></a><br>
#### DAG Tasks - `Operators`
Airflow Operators are used in creating individual tasks in the DAG. Properties specific to a task will be configured in the operator.

<h6>Some of the most commonly used Airflow Operators</h6>  
`PythonOperator` - Calls a python function. An alternative way to create a python function task is to use TaskFlow API which is available from Airflow 2.0
`BashOperator` - Executes a UNIX command

<a name = "oprcd"></a>
```py
#Operators
# Bash operator tasks that executes unix command - echo
task_1 = BashOperator(
    task_id='first_task',
    bash_command='echo "First Task"',
    dag=my_workflow_dag,
)

# Python Operator that prints the details of current job using context variable
 task_2 = PythonOperator(
    task_id='second_task',
    python_callable=print_fn,
    provide_context=True,
    on_success_callback = task_success_alert,
    dag=my_workflow_dag,
)

#Python Function

def print_fn(ds):
    """
    Implement your python task logic here.
    """
    #Prints job details passed through provide_context parameter
    print(ds)
```

Now that we have our DAG ready, let's see where I have configured the features in it.

<a name = "retry"></a><h4>✨ RETRY</h4>
Airflow handles errors and failures gracefully.
If you prefer to re-run a failed task multiple times before aborting the workflow run, use [**retries**](#def) argument. So that an erroneous task will be executed up to the defined number of times before it's marked as failed.

<a name = "alert"></a><h4>✨ ALERTS</h4>
[**on_success_callback**](#oprcd) and [**on_failure_callback**](#defcd) arguments are used to trigger some actions once the workflow succeeds or fails respectively. This will be useful to send personalized alerts to internal team via Slack, Email, or any other API call when a workflow task succeeds or fails.

###### 👇 Sample <i>Slack notification</i> received from Airflow
![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/wyaq96ipep80t560o5yd.jpg)
###### View Code - [Success](https://github.com/Sri-nidhi/Airflow-Notes/blob/f03ba5033b1a7e5c1571c5e8edef969fbd65ae76/utils/alert.py#L11), [Failure](https://github.com/Sri-nidhi/Airflow-Notes/blob/f03ba5033b1a7e5c1571c5e8edef969fbd65ae76/utils/alert.py#L30)

<a name = "web"></a><h4>✨ WEBUI </h4>
Executing `airflow webserver` in CLI starts airflow service which opens up Airflow WebUI in  `0.0.0.0::8080` URL. 

Since Airflow is made of Flask framework, you can even extend WebUI by creating additional pages using [Flask AppBuilder](https://www.coditation.com/airflow-ui-plugin-development-a-walkthrough/).


<a name = "sch"></a><h4>✨ SCHEDULING</h4>
Scheduler service - the heart of Airflow, needs to be <i>Up and Running</i> to run the DAGS. 
Once Airflow is installed, execute `airflow scheduler` to start scheduler service.

![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/08jet45zmbjgtwfyfi51.png)

In addition to that, DAGs can be made to run automatically at a particular time by providing a <a href="https://crontab.guru/" target="_blank">cron notation</a> in [`schedule_interval`](#dagcd)


<a name = "api"></a><h4>✨ API</h4>
Airflow 2.0 has released Airflow API that opens the door to Automation with Airflow. With Airflow API, one can automate DAG triggers, reads, and other [operations](https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html) that are possible using WebUI.
![Alt Text](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/dkyyn19zsnirqg7izov5.png)

------------

### **Setting up your Airflow Instance**
Here are few great resources :
[On Windows with Docker](https://dev.to/jfhbrook/how-to-run-airflow-on-windows-with-docker-2d01)
[On Ubuntu Local Setup](https://medium.com/@taufiq_ibrahim/apache-airflow-installation-on-ubuntu-ddc087482c14)
[On Kubernetes](https://github.com/rolanddb/airflow-on-kubernetes)
[AWS Using cloudformation](https://github.com/villasv/aws-airflow-stack)


If you don't have the bandwidth to set up and maintain airflow in your own infrastructure, here are the commercial Airflow-as-a-service providers:

[Google Cloud Composer](https://cloud.google.com/composer/) - managed service built atop Google Cloud and Airflow.
[Astronomer.io](https://www.astronomer.io/) - In addition to hosting Airflow in their infrastructure, they provide solutions focused on airflow and support services.

------------

### Is Airflow the only option for workflows?
<i>Certainly Not</i>. It depends on different factors.
Say your entire product is hosted in a single cloud provider ( AWS / Azure/ GCP) and you are fine with vendor lock-in. 
Then,
- For AWS, [AWS step function](https://aws.amazon.com/step-functions/) will be a good option. 
- For Azure, you can opt for [Azure Data Factory](https://azure.microsoft.com/en-in/services/data-factory/). 
- For GCP, [Google Cloud Composer](https://cloud.google.com/composer/) will be the best fit.

Whereas if you want to host your product in multi-cloud - 
Airflow will be a  better fit. There are other workflows in the market similar to Airflow :
[Prefect](https://www.prefect.io/)
[Luigi](https://github.com/spotify/luigi)
[Dagster](https://docs.DAGster.io/) 

------------

### **Airflow Best practices / Tips:**
1. Once you have airflow installed, modify the configurations of your airflow instance in the master config file - **[airflow.cfg](https://github.com/apache/airflow/blob/master/airflow/config_templates/default_airflow.cfg)**
2. Avoid creating complex data processing logic in Airflow, as airflow is not intended to work as a data processing engine. Its focus is to orchestrate data flow jobs. (For data processing, Run the processing script in a batch compute or spark framework and invoke from airflow)
3. Use [macros and templates](https://marclamberti.com/blog/templates-macros-apache-airflow/) to avoid hard coding values.
4. While migrating the airflow metadata db to a new database, use <i>"airflow upgradedb" (For 2.0 airflow db upgrade)</i> instead of <i>"airflow initdb" (For 2.0 airflow db init)</i>
5.  When you want to implement a task out of the box, create [custom operators](https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html) using the `BaseOperator` Class for re-using them.
6. Use [Airflow Plugins](https://github.com/airflow-plugins/), to connect to third-party tools/data sources.
7. Use Postgres database as airflow's metadata DB to use Local and CeleryExecutors.

If you have any queries or any more tips that you think might be useful, comment below on the post.


Going forward, I'll publish detailed posts on tools and frameworks used by Data Engineers day in and day out. 



<img src="https://media.giphy.com/media/J4zA6LplubvC5weDyo/giphy.gif" width="120" height="120"/>
Follow for updates.