Apache Airflow is an open-source platform for programmatically authoring, scheduling, and monitoring workflows. It was built for increasingly complex ETL and data pipelines, but its extensibility, production quality, and GUI support support broader domains, including deep learning. section 9.3.1
Build an Airflow workflow around a Python-defined directed acyclic graph (DAG):
- Define the DAG and its tasks, then declare task dependencies. section 9.3.1
- Implement tasks with operators, using predefined or community-managed operators such as MySqlOperator, SimpleHttpOperator, and Docker operator. section 9.3.1
- Use PythonOperator for customized Python functions and dynamic workflows when tasks must be created programmatically. section 9.3.1
- Inspect deployed workflow execution through the web UI or CLI commands that list DAGs, tasks, and task trees. section 9.3.1
Daily data-processing workflow
A representative DAG runs daily, waits for data.csv with FileSensor, transforms the data with PythonOperator, creates a database table with PostgresOperator, and continues with dependent processing tasks. section 9.3 section 9.3.1
Listing 9.1 A sample Airflow workflow definition
# declare the workflow DAG.
with DAG(dag_id="data_process_dag",
schedule_interval="@daily",
default_args=default_args,
template_searchpath=[f"{os.environ['AIRFLOW_HOME']}"],
catchup=False) as dag:
# define tasks of the workflow, each code section below is a task
is_new_data_available = FileSensor( #1
task_id="is_new_data_available",
fs_conn_id="data_path",
filepath="data.csv",
.. .. ..
)
# define data transformation task
transform_data = PythonOperator(
task_id="transform_data",
python_callable=transform_data #2
)
# define table creation task
create_table = PostgresOperator( #3
task_id="create_table",
sql='''CREATE TABLE IF NOT EXISTS invoices (
.. .. ..
);''',
postgres_conn_id='postgres',
database='customer_data'
)
save_into_db = PythonOperator(
task_id='save_into_db',
python_callable=store_in_db
)
notify_data_science_team = SlackWebhookOperator(
task_id='notify_data_science_team',
http_conn_id='slack_conn',
webhook_token=slack_token,
message="Data Science Notification \n"
.. .. ..
)
# Step two, declare task dependencies in the workflow
is_new_data_available >> transform_data
transform_data >> create_table >> save_into_db
save_into_db >> notify_data_science_team
save_into_db >> create_report
# The actual data transformation logic, which is referenced
# in the “transform_data” task.
def transform_data(*args, **kwargs):
.. .. ..
Inspect the sample Python DAG definition, including its daily schedule, sensor, transformation task, and table-creation task. section 9.3.1
