AWS Managed Airflow for your complex workflows
Tags: aws, airflow, python, mlops
Published: 2021-07-18 01:43:38

**Airflow** is a tool to orchestrate complex workflow which was created at Airbnb in 2014. Airflow gained significant traction across several organizations in recent days due to the ability to create complex data pipelines with ease. The Airflow user interface (UI) serves as an operational dashboard to schedule, monitor and control any scripts or applications. 

Despite Airflow has been adopted by several organizations, deploying and managing the infrastructure has always been challenging and introduces operational overhead. To combat this problem many companies came up with the idea of introducing managed Airflow Infrastructure such as MWAA from Amazon Web Services (AWS). AWS takes the responsibility for keeping your infrastructure with almost no downtime and up-to-date security patches in place along with the latest version readily available to use. 

Besides, the other key benefits of using AWS MWAA is its elasticity, the ability to scale up and down based on the workload, and the easiness of building and deploying production-grade secure infrastructure with seamless integration with other AWS Services. 

## Airflow Basics 

Let's dive into some key concepts of Airflow :)

In Airflow, workflows are defined in a Python file which is also referred to as DAG. You can imagine a DAG as a single job that can have multiple tasks in general. There are three common parts in every DAG those are as follows:

- DAG Initialization 
- Tasks 
- Tasks Dependencies 

### DAG (Directed Acyclic Graphs)

DAGs are written in Python and are often identified by their unique `dag_id`. During the initialization, we specify when to start, scheduled time and so forth. Here is a simple DAG below: 

```python
from airflow.models import DAG
from airflow.utils.dates import days_ago

dag = DAG(
          dag_id="sample_dag",
          start_date=days_ago(2), 
          description="Sample DAG",
          schedule_interval='@daily')
```

### Task 

Tasks perform different actions from executing a piece of Shell script to triggering EMR jobs. It is necessary to have a DAG before we create any task. Also, every task in a DAG is defined by an operator and similar to `dag_id`, the `task_id` would need to be unique within the DAG. 

```
def function_a (**kwargs):
        name = kwargs['name']
        return f'hello {name} !!'

first_task = PythonOperator(
        task_id="first_task", 
        python_callable= function_a,
        op_kwargs= {'name': 'Fayaz'}, 
        dag= dag)

second_task = DummyOperator(task_id="second_task", dag=dag)
```

### Task Dependencies 

Now the last part of the DAG is to create dependencies among the tasks. In this case, we are going to trigger the `first_task` first and then trigger the `second_task` as soon as the first_task completes. So it will look like this:

```
first_task >> second_task
```

Now we understood what is Airflow and how to create a simple DAG so let's spin up the AWS MWAA to run this DAG. 

> *You need to use your AWS Account to perform the next few steps which may incur some charges.*

## Setting up the Managed Airflow Instance in AWS

Before we create a new MWAA environment, we would need to create an `S3 bucket` which must have versions enabled. 

**Step 1.** Go to [Managed Airflow Console](https://console.aws.amazon.com/mwaa/home?#create/environment) and click Create Environment
 
![2](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/3fbj7j107m94ou85l42c.PNG)
 
**Step 2.** Enter a **name** and choose the Airflow version as **2.0.2(Latest)**

![3](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/y4pu7o86x8vpu1v8pv7j.PNG)

**Step 3.** Choose the S3 bucket - the one you have created 
**Step 4.** For Dags folder type **s3://{your-bucket-name}/dags**
**Step 5.** Click **Next** 

![5](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/76re3ydkbtxsudtf2a64.PNG)

**Step 6.** Click on **Create MWAA VPC**

![6](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ykk0gfszywk1h6ff5k1k.PNG)

**Step 7.** It will take you to a page with a bunch of VPC, Subnets details. Click **Create Stack** which may take a few mins to complete

![7](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/dzmx0kyzz6vtuk3uadrn.PNG)

**Step 8.** Choose the VPC you just created and scroll down to enter environment class and other configurations

![8](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/7dkbp4p132qou378blun.PNG)

**Step 9.**  Choose **Create a new role** and click Next

![9](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/0n7owa5xiwjdx17n45gm.PNG)

**Step 10.** Verify all the details and click **Create environment**

![10](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/me0syi4hg0zj5oq7ki0n.PNG)
 
Generally, it takes 10-20 mins to spin up the Airflow Infrastructure so this is the time to get your coffee ☕ before we deploy our very first DAG 🤣😂 

🎉🎉🎉 When you refresh in a few minutes, you will see the environment status as `Available` so click **Open Airflow UI**

![11](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/iz8gcisbyxxd99qqtlkg.PNG)

![12](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/2kuual4eq4zd8h990kwm.PNG)

Yay!! Now we got our environment up and ready to go. So let's deploy our first DAG. 

## Deploying DAGs in the AWS MWAA 

Step 1: To deploy the DAG, we would need to copy the `.py` file to our `s3/dags` location. Copy the below code and put that in a `.py` file and save it as demo_dag.py in your local. 

```
"""
Importing necessary modules
"""
from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator


"""
Initializing DAGs
"""
dag = DAG(
          dag_id="grepy_sample_dag",
          start_date=days_ago(2), 
          description="DAG which orchestrates a simple ML workflow",
          schedule_interval='@daily')

"""
Creating Tasks
"""
def function_a (**kwargs):
        name = kwargs['name']
        return f'hello {name} !!'

first_task = PythonOperator(
        task_id="first_task", 
        python_callable= function_a,
        op_kwargs= {'name': 'Fayaz'}, 
        dag= dag)

second_task = DummyOperator(task_id="second_task", dag=dag)

"""
Dependencies
"""
first_task >> second_task

```
Step 2: Upload the `demo_dag.py` file to your s3/dags folder. 

![13](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/crq0xbza1gw42khorhk0.PNG)

Step 3: That's it!! Now again it may take a few minutes when you deploy a DAG for the first time, but it will look like this 

![14](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/d1w73kw017364fr44n2e.PNG)

When you toggle on for the first time, the DAG will automatically be triggered so click on the DAG name which will take you to the tree view where you see the job status and task dependencies. 

One of the best parts is you can see all the logs when you go to the Graph view which is also available in CloudWatch so you can ingest that to your Splunk or ELK for further analysis. 

![15](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/fuw34qgavjfd1rcfwl4g.PNG)

## Conclusion 

In this post, we took a high-level look at Airflow basics and we dived into AWS Managed Airflow along with a sample DAG deployment which runs a simple Python function. Similarly, you can orchestrate any type of task using various operators which is available on any Airflow infrastructure.

## For further reading 

-  https://aws.amazon.com/managed-workflows-for-apache-airflow/getting-started/ 


Thank you for your time and happy learning !! 😊