2021-03-08 TIL March 08, 2021

from airflow.models import DAG
from airflow.operators.mysql_operator import MySqlOperator
from airflow.providers.http.sensors.http import HttpSensor
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.operators.python import PythonOperator

from datetime import datetime
from sqlalchemy import create_engine
import json
import os
import pandas as pd

default_args = {
    'start_date': datetime(2021, 1, 1),
}


def _processing_user(ti):
    users = ti.xcom_pull(task_ids=['extracting_user'])
    if not len(users) or 'results' not in users[0]:
        raise ValueError('User is empty')
    user = users[0]['results'][0]
    processed_user = pd.json_normalize({
        'firstname': user['name']['first'],
        'lastname': user['name']['last'],
        'country': user['location']['country'],
        'username': user['login']['username'],
        'password': user['login']['password'],
        'email': user['email']
    })
    processed_user.to_csv('/tmp/processed_user.csv', index=None, header=False)


def _csvToSql():
    try:
        engine = create_engine("mysql+pymysql://{user}:{password}@{host}/{database}"
                               .format(user=os.environ["external_raw_data_mysql_user"],
                                       password=os.environ["external_raw_data_mysql_password"],
                                       host=os.environ["external_raw_data_mysql_host"],
                                       database="airflow_test"),
                               encoding='utf-8')
        conn = engine.connect()
    except:
        print('Cannot connect.')

    df = pd.read_csv('/tmp/processed_user.csv',
                     names=['firstname', 'lastname',
                            'country', 'username', 'password', 'email'])
    print(df.head())
    df.to_sql(name='users', con=engine, if_exists='append', index=False)

    conn.close()

    return 'Read .csv and written to the MySQL database'


with DAG('user_processing', schedule_interval='@daily',
         default_args=default_args,
         catchup=False) as dag:
    # define tasks/operators
    create_table = MySqlOperator(
        task_id='creating_table',
        mysql_conn_id='local_mysql_default',
        sql='''
            CREATE TABLE IF NOT EXISTS users (
                email VARCHAR(255) NOT NULL,
                firstname TEXT NOT NULL,
                lastname TEXT NOT NULL,
                country TEXT NOT NULL,
                username TEXT NOT NULL,
                password TEXT NOT NULL,
                PRIMARY KEY (email)
            );
        ''',
    )

    is_api_available = HttpSensor(
        task_id='is_api_available',
        http_conn_id='user_api',
        endpoint='api/'
    )

    extracting_user = SimpleHttpOperator(
        task_id='extracting_user',
        http_conn_id='user_api',
        endpoint='api/',
        method='GET',
        response_filter=lambda response: json.loads(response.text),
        log_response=True
    )

    processing_user = PythonOperator(
        task_id='processing_user',
        python_callable=_processing_user
    )

    storing_user = PythonOperator(
        task_id='storing_user',
        python_callable=_csvToSql
    )

create_table >> is_api_available >> extracting_user >> processing_user >> storing_user

from airflow import DAG
from airflow.operators.bash import BashOperator

from datetime import datetime

default_args = {
    'start_date': datetime(2021, 1, 1)
}

with DAG('parallel_dag', schedule_interval='@daily', default_args=default_args, catchup=False) as dag:

    task_1 = BashOperator(
        task_id='task_1',
        bash_command='sleep 3'
    )

    task_2 = BashOperator(
        task_id='task_2',
        bash_command='sleep 3'
    )

    task_3 = BashOperator(
        task_id='task_3',
        bash_command='sleep 3'
    )

    task_4 = BashOperator(
        task_id='task_4',
        bash_command='sleep 3'
    )

    task_1 >> [task_2, task_3] >> task_4

pip install 'apache-airflow[mysql]'

sql_alchemy_conn = mysql://[DB_USER]:[DB_PASSWORD]@127.0.0.1:3306/[DB_DATABASE]

executor = LocalExecutor