Learn how to Create, Optimize, and Deploy Data Pipelines using Apache Airflow with Practical Examples and GitHub Resources and a Comprehensive Guide to Building Efficient ETL Solutions

What is a Data Pipeline?

A data pipeline is simply, a series of steps in which data is processed

Data Partitioning

• Pipeline data partitioning is the process of isolating data to be analyzed by one or more attributes, such as time, logical type or data size
• Data partitioning often leads to faster and more reliable pipelines with Apache Airflow
• Types of Data Partitioning:

Schedule partitioning

• Not only are schedules great for reducing the amount of data our pipelines have to process, but they also help us guarantee that we can meet timing guarantees that our data consumers may need

Logical partitioning

• Conceptually related data can be partitioned into discrete segments and processed separately. This process of separating data based on its conceptual relationship is called logical partitioning.
• With logical partitioning, unrelated things belong in separate steps. Consider your dependencies and separate processing around those boundaries
• Examples of such partitioning are by date and time

Size partitioning

• Size partitioning separates data for processing based on desired or required storage limits
• This essentially sets the amount of data included in a data pipeline run

Why partition data?

• Pipelines designed to work with partitioned data fail more gracefully. Smaller datasets, smaller time periods, and related concepts are easier to debug than big datasets, large time periods, and unrelated concepts of Apache Airflow
• If data is partitioned appropriately, tasks will naturally have fewer dependencies on each other
• Airflow will be able to parallelize execution of DAGs to produce results even faster

Data Validation

Data Validation is the process of ensuring that data is present, correct & meaningful. Ensuring the quality of data through automated validation checks is a critical step in building data pipelines at any organization

Data Quality

• Data Quality is a measure of how well a dataset satisfies its intended use
• Examples of Data Quality Requirements
• Data must be a certain size of apache airflow
• Data must be accurate to some margin of error
• Data must arrive within a given timeframe from the start of execution
• Pipelines must run on a particular schedule
• Data must not contain any sensitive information

Directed Acyclic Graphs

• Directed Acyclic Graphs (DAGs): DAGs are a special subset of graphs in which the edges between nodes have a specific direction, and no cycles exist.

Apache Airflow

• What is Apache Airflow ?
• Airflow is a platform to programmatically author, schedule and monitor workflows

• Use airflow to author workflows as directed acyclic graphs (DAGs) of tasks
• The airflow scheduler executes your tasks on an array of workers while following the specified dependencies
• When workflows are defined as code, they become more maintainable, versionable, testable, and collaborative
• Airflow concepts (Taken from Apache Airflow documentation)

Operators

• Operators determine what actually gets done by a task. An operator describes a single task in a workflow. Operators are usually (but not always) atomic. The DAG will make sure that operators run in the correct order; other than those dependencies, operators generally run independently

Tasks

• Once an operator is instantiated, it is referred to as a “task” The instantiation defines specific values when calling the abstract operator, and the parameterized task becomes a node in a DAG
• A task instance represents a specific run of a task and is characterized as the combination of a DAG, a task, and a point in time. Task instances also have an indicative state, which could be “running”, “success”, “failed”, “skipped”, “up for retry”, etc

DAGs

• In Apache Airflow, a DAG — or a Directed Acyclic Graph — is a collection of all the tasks you want to run, organized in a way that reflects their relationships and dependencies
• A DAG run is a physical instance of a DAG, containing task instances that run for a specific execution_date. A DAG run is usually created by the Airflow scheduler, but can also be created by an external trigger

Hooks

• Hooks are interfaces to external platforms and databases like Hive, S3, MySQL, Postgres, HDFS, and Pig. Hooks implement a common interface when possible, and act as a building block for operators
• They also use the airflow.models.connection.Connection model to retrieve hostnames and authentication information
• Hooks keep authentication code and information out of pipelines, centralized in the metadata database

Connections

• The information needed to connect to external systems is stored in the Apache Airflow metastore database and can be managed in the UI (Menu -> Admin -> Connections)
• A conn_id is defined there, and hostname / login / password / schema information attached to it
• Airflow pipelines retrieve centrally-managed connections information by specifying the relevant conn_id

Variables

• Variables are a generic way to store and retrieve arbitrary content or settings as a simple key value store within Airflow
• Variables can be listed, created, updated and deleted from the UI (Admin -> Variables), code or CLI. In addition, json settings files can be bulk uploaded through the UI

Context & Templating

• Apache Airflow leverages templating to allow users to “fill in the blank” with important runtime variables for tasks
• See: https://airflow.apache.org/docs/stable/macros-ref for a list of context variables

Airflow Plugins

• Apache Airflow was built with the intention of allowing its users to extend and customize its functionality through plugins.
• The most common types of user-created plugins for Apache Airflow are Operators and Hooks. These plugins make DAGs reusable and simpler to maintain

To create custom operator, follow the steps

1. Identify Operators that perform similar functions and can be consolidated
2. Define a new Operator in the plugins folder
3. Replace the original Operators with your new custom one, re-parameterize, and instantiate them

Airflow subdags

• Commonly repeated series of tasks within DAGs can be captured as reusable SubDAGs
• Benefits include:
• Decrease the amount of code we need to write and maintain to create a new DAG
• Easier to understand the high level goals of a DAG
• Bug fixes, speedups, and other enhancements can be made more quickly and distributed to all DAGs that use that SubDAG

Drawbacks of Using SubDAGs:

• Limit the visibility within the Airflow UI
• Abstraction makes understanding what the DAG is doing more difficult
• Encourages premature optimization

Monitoring

• Apache airflow can surface metrics and emails to help you stay on top of pipeline issues
• SLAs
• Airflow DAGs may optionally specify an SLA, or “Service Level Agreement”, which is defined as a time by which a DAG must complete
• For time-sensitive applications these features are critical for developing trust amongst pipeline customers and ensuring that data is delivered while it is still meaningful

Emails and Alerts

• Apache Airflow can be configured to send emails on DAG and task state changes
• These state changes may include successes, failures, or retries
• Failure emails can easily trigger alerts

Metrics

• Apache airflow comes out of the box with the ability to send system metrics using a metrics aggregator called stats
• Statsd can be coupled with metrics visualization tools like Grafana to provide high level insights into the overall performance of DAGs, jobs, and tasks

Best practices for data pipelining

• Task Boundaries
  DAG tasks should be designed such that they are:
• Atomic and have a single purpose
• Maximize parallelism
• Make failure states obvious

Let’s Prectice of Airflow on all above concept’s

Hello Apache Airflow

Instructions
Define a function that uses the python logger to log a function. Then finish filling in the details of the DAG down below. Once you’ve done that, run “/opt/airflow/start.sh” command to start the web server. Once the Airflow web server is ready, open the Airflow UI using the “Access Airflow” button. Turn your DAG “On”, and then Run your DAG. If you get stuck, you can take a look at the solution file or the video walkthrough on the next page.

import datetime
import logging

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

def my_function():
logging.info("hello airflow")

dag = DAG(
'mock_airflow_dag',
start_date=datetime.datetime.now())

greet_task = PythonOperator(
task_id="hello_airflow_task",
python_callable=my_function,
dag=dag
)

Context And Templating

Instructions
Use the Airflow context in the pythonoperator to complete the TODOs below. Once you are done, run your DAG and check the logs to see the context in use.

import datetime
import logging

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python_operator import PythonOperator
from airflow.hooks.S3_hook import S3Hook

def log_details(*args, **kwargs):
#
# TODO: Extract ds, run_id, prev_ds, and next_ds from the kwargs, and log them
# NOTE: Look here for context variables passed in on kwargs:
# https://airflow.apache.org/macros.html
#
ds = kwargs['ds']
run_id = kwargs['run_id']
previous_ds = kwargs['prev_ds']
next_ds = kwargs['next_ds']

logging.info(f"Execution date is {ds}")
logging.info(f"My run id is {run_id}")
if previous_ds:
logging.info(f"My previous run was on {previous_ds}")
if next_ds:
logging.info(f"My next run will be {next_ds}")

dag = DAG(
'lesson1.exercise5',
schedule_interval="@daily",
start_date=datetime.datetime.now() - datetime.timedelta(days=2)
)

list_task = PythonOperator(
task_id="log_details",
python_callable=log_details,
provide_context=True,
dag=dag
)

Dag for Subdag

Consolidate check_trips and check_stations into a single check in the subdag
As we did with the create and copy in the demoimport datetime

from airflow import DAG
from airflow.operators.postgres_operator import PostgresOperator
from airflow.operators.subdag_operator import SubDagOperator
from airflow.operators.udacity_plugin import HasRowsOperator

from lesson3.exercise3.subdag import get_s3_to_redshift_dag
import sql_statements

start_date = datetime.datetime.utcnow()

dag = DAG(
"lesson3.exercise3",
start_date=start_date,
)

trips_task_id = "trips_subdag"
trips_subdag_task = SubDagOperator(
subdag=get_s3_to_redshift_dag(
"lesson3.exercise3",
trips_task_id,
"redshift",
"aws_credentials",
"trips",
sql_statements.CREATE_TRIPS_TABLE_SQL,
s3_bucket="udac-data-pipelines",
s3_key="divvy/unpartitioned/divvy_trips_2018.csv",
start_date=start_date,
),
task_id=trips_task_id,
dag=dag,
)

stations_task_id = "stations_subdag"
stations_subdag_task = SubDagOperator(
subdag=get_s3_to_redshift_dag(
"lesson3.exercise3",
stations_task_id,
"redshift",
"aws_credentials",
"stations",
sql_statements.CREATE_STATIONS_TABLE_SQL,
s3_bucket="udac-data-pipelines",
s3_key="divvy/unpartitioned/divvy_stations_2017.csv",
start_date=start_date,
),
task_id=stations_task_id,
dag=dag,
)

#

check_trips = HasRowsOperator(
task_id="check_trips_data",
dag=dag,
redshift_conn_id="redshift",
table="trips"
)

check_stations = HasRowsOperator(
task_id="check_stations_data",
dag=dag,
redshift_conn_id="redshift",
table="stations"
)

location_traffic_task = PostgresOperator(
task_id="calculate_location_traffic",
dag=dag,
postgres_conn_id="redshift",
sql=sql_statements.LOCATION_TRAFFIC_SQL
)

All These topic covered in this article:

Data Pipelines, ETL solutions, ApacheAirflow, DataPartitioning, DataValidation, DirectedAcyclicGraphs, AirflowOperators, DAGs, AirflowHooks, AirflowConnections, AirflowVariables, AirflowSubDAGs, AirflowMonitoring, SLAinAirflow, DataQuality, ETLBestPractices, DataPipelineOptimization, TaskBoundaries, GrafanaMetrics, DataEngineering

🚀 Connect with Me and Explore My World! 🌍

Ready to dive into an exciting world of knowledge and collaboration? Join me on various platforms, and let’s embark on an incredible journey together!

💼 LinkedIn: Discover my professional endeavors and join an ever-growing network of like-minded individuals. Connect now!
🔬 Kaggle: Let’s explore, analyze, and conquer the thrilling world of data science and machine learning together! I am a Kaggle 2x Master. Kaggle with me!
💻 GitHub: Explore my coding playground, where I build fascinating projects and contribute to the open-source community. Check out my repositories!

Don’t miss this opportunity to connect with a passionate and curious mind like mine. Let’s learn, grow, and make a positive impact on the world together! See you there! 😊 (Arslan Ali Kaggle Master)