Orchestrate ETL Pipelines With Apache Airflow®

This document demonstrates a simple extract, transform, and load (ETL) pipeline orchestration with Apache Airflow.

The document is intended for data engineers who have a basic understanding of the following:

• ETL processes.
• Python / pandas.
• PostgreSQL®.
• Windows Subsystem for Linux (WSL 2).
• Ubuntu.
• Transmission Control Protocol and Internet Protocol (TCP/IP).

The document focuses on orchestration and does not explain the implementation of the ETL processes.

Open-Meteo Weather API ETL pipeline orchestration

Problem Definition

In my previous case studies, I built data pipelines that required manual execution. The absence of orchestration introduces repetitive tasks, and limits scalability. As the system grows, and data processing requirements increase, manual execution becomes impractical, and leads to operational delays.

In this case study, I orchestrate the Open-Meteo Weather API ETL pipeline. API data are offered under Attribution 4.0 International (CC BY 4.0).

This forecast ETL pipeline monitors daily Weather variables for the Casablanca metropolitan area.

The data pipeline workflow is defined as follows:

1. The pipeline retrieves a selected subset of Weather variables from the API using predefined parameters.
2. The pipeline flattens the required nested objects from the JSON response into DataFrames. This process generates two DataFrames. The first one stores daily Weather data for the area at two-hour intervals. The second one tracks the daily measurement units and their insertion date.
3. The pipeline loads the resulting data into two distinct relational tables in a PostgreSQL database.

For the full implementation of the ETL processes, see the GitHub repository.

Problem Solution

Modern data engineering offers a wide range of orchestration tools for batch processing and stream processing.

Apache Airflow® (or simply Airflow) is an open-source orchestration platform. It enables users to write, schedule, and monitor data workflows. Airflow supports batch processing, which makes the platform suitable for orchestrating the Open-Meteo Weather API ETL pipeline.

Implementation

Execution Environment

Figure 1 provides an overview of the pipeline architecture and the Airflow orchestration implementation.

Figure 1. Overview of the Open-Meteo API ETL pipeline architecture and the implementation of the Airflow orchestration.

WSL 2 provides an Ubuntu environment for installing and running Airflow, limiting the risk of compatibility issues. The Python project also runs inside WSL 2, while the PostgreSQL database runs locally on the host Windows operating system.

Airflow Installation

In this case study, I installed Airflow in two environments:

• A main Airflow installation in the Ubuntu WSL 2 environment.
• An Airflow installation inside the Python application's virtual environment using pip.

The main Airflow installation is responsible for the Airflow environment and its associated configuration, including the metadata database, administrator credentials, and other Airflow settings.

The installation inside the Python application's virtual environment provides the Airflow packages required by the project and allows Airflow to run within the same Python environment as the application's dependencies.

This distinction matters later when running Airflow standalone.

For further information on installing Airflow on WSL 2, see the official documentation.

Airflow logic

Airflow manages the workflow execution, while the Python application executes the ETL processes.

In Airflow, a Directed Acyclic Graph (DAG) is a file that defines the structure of workflows. A DAG contains all the tasks and their dependencies, which determine the order in which Airflow orchestrates the pipeline.

In order to orchestrate this ETL pipeline with Airflow, we define the following steps:

1. Create a DAG file.
2. Define the tasks and dependencies.
3. Configure the TCP/IP connection between the Python application and the PostgreSQL database.
4. Verify the creation of the DAG using the Airflow standalone GUI.
5. Monitor the data loading in the PostgreSQL database.

1. Create the DAG file

The Airflow project uses the airflow/dags directory in the root of the Python project. The dags subfolder contains all the DAG files that Airflow can use. In this case, I define a single file.

The project structure is as follows:

open_meteo_airflow_pipeline/
│
├── airflow/
│ └── dags/
├── pipeline/
│ ├── extract.py
│ ├── transform.py
│ └── load.py
├── main.py
├── .env
├── requirements.txt
└── README.md

Inside dags, I create the open_meteo_etl_dag.py DAG file.

2. Define the tasks and dependencies

The following code is defined inside open_meteo_etl_dag.py:

import sys
import os

import pandas as pd

sys.path.append(
    os.path.abspath(
        os.path.join(os.path.dirname(__file__), "../..")
    )
)

from datetime import datetime, timedelta

from airflow.sdk import DAG, task

from pipeline.extract import fetch_data
from pipeline.transform import transform_data
from pipeline.load import load_data

default_args = {
    "owner": "kitchen_code",
    "retries": 5,
    "retry_delay": timedelta(minutes=3),
    "execution_timeout": timedelta(minutes=15),
}

with DAG(
    dag_id="etl_dag",
    start_date=datetime(2026, 8, 3),
    schedule="@daily",
    default_args=default_args,
    catchup=False
) as dag:

    @task
    def extract_task():
        return fetch_data()

    @task
    def transform_task(extracted_data):
        units, latitude, longitude, hourly_data = extracted_data
        units_df, weather_df = transform_data(
            units,
            latitude,
            longitude,
            hourly_data
        )

        units_path = "/tmp/weather_units.csv"
        weather_path = "/tmp/weather_data.csv"

        units_df.to_csv(units_path, index=False)
        weather_df.to_csv(weather_path, index=False)

        return units_path, weather_path


    @task
    def load_task(paths):
        units_path, weather_path = paths
        units_df = pd.read_csv(units_path)
        weather_df = pd.read_csv(weather_path)

        load_data(units_df, weather_df)


    extracted_data = extract_task()
    transformed_data = transform_task(extracted_data)
    load_task(transformed_data)

First, I import the Airflow components to define the DAG and the tasks:

from airflow.sdk import DAG, task

The DAG constructor accepts a wide range of parameters. In this file, I define the following parameters:

• dag_id: defines the unique identifier that Airflow uses to identify the DAG.
• start_date: defines the date from which the DAG's scheduled runs should begin.
• schedule: defines the frequency at which Airflow schedules the DAG. In this project, the DAG is configured to run once per day using @daily.
• default_args: defines a Python dictionary containing default configuration parameters. In this project, it contains: owner: identifies the owner responsible for this DAG. retries: defines the number of times Airflow should retry a task after a failure. retry_delay: defines the amount of time Airflow should wait before retrying to run a task. execution_timeout: defines the maximum amount of time a task is allowed to run before Airflow stops it due to a timeout.
• catchup: tells a DAG whether to run all missed time intervals that occurred between its start_date and the current time.

Once I define the DAG parameters, I create the tasks. In this context, we need three tasks: one for each ETL process.

I use the @task decorator to create the first task, extract_task, for the extraction process:

@task
def extract_task():
    return fetch_data()

I import the function fetch_data() from the extract.py module, and I return the result generated by the function from the task.

Using @task, I create the second task transform_task for the transformation process:

 @task
    def transform_task(extracted_data):
        units, latitude, longitude, hourly_data = extracted_data
        units_df, weather_df = transform_data(
            units,
            latitude,
            longitude,
            hourly_data
        )

        units_path = "/tmp/weather_units.csv"
        weather_path = "/tmp/weather_data.csv"

        units_df.to_csv(units_path, index=False)
        weather_df.to_csv(weather_path, index=False)

        return units_path, weather_path

The tasks are connected as follows:

extracted_data = extract_task()
transformed_data = transform_task(extracted_data)

Since the tasks were created using the @task decorator, Airflow automatically handles the communication between them using the XComs.

Unlike the traditional approach, where XComs are managed with xcom_push() and xcom_pull(), the TaskFlow API handles this process automatically through task inputs and outputs. This defines the dependencies between tasks.

The returned values from extract_task are made available to transform_task through the extracted_data parameter. The transform_task is defined as follows:

1. The task receives the values returned by fetch_data(): units, latitude, longitude, hourly_data.
2. The task imports the transform_data() function from the transform.py module and passes the returned values of fetch_data() to it. The end result is two DataFrames: units_df and weather_df.
3. The task converts the two DataFrames into CSV files. This approach allows the next task to access the transformed data without passing the entire DataFrames through XCom. At last, the task returns the path of each file.

Using @task, I create the last task load_task, for the loading process:

    @task
    def load_task(paths):
        units_path, weather_path = paths
        units_df = pd.read_csv(units_path)
        weather_df = pd.read_csv(weather_path)

        load_data(units_df, weather_df)

The tasks are connected as follows:

transformed_data = transform_task(extracted_data)
load_task(transformed_data)

The load_task takes paths as a parameter. This allows the task to read the CSV files using their paths, and load their contents back into pandas DataFrame.

The task calls the load_data() function from the load.py module. This function executes the following operations:

1. Establishes a connection to the PostgreSQL database using the psycopg driver.
2. Executes Data Definition Language statements to create the weather_data, and weather_units tables.
3. Executes Data Manipulation Language statements to insert the transformed data into the tables.

3. Configure the TCP/IP connection between the Python application and the PostgreSQL database.

The Python application runs inside WSL 2, while the PostgreSQL database runs locally on the Windows host operating system. Since the application and the database run in different environments, the connection between them is established by TCP/IP.

The psycopg driver requires several connection parameters. These parameters are stored in environment variables:

DB_NAME=
DB_USER=
DB_PASSWORD=
DB_HOST=
DB_PORT=

In my case, the main configuration issue concerns the DB_HOST parameter. Since the Python application runs inside WSL 2, the PostgreSQL server that runs on the host operating system cannot be reached using the default localhost address.

Therefore, I use the IP address of the Windows host as it is accessible from the WSL 2 environment.

To retrieve this IP address, run the following command in your Windows terminal:

ipconfig

Look for the IP address under the vEthernet (WSL (Hyper-V firewall)) network adapter. This is the IP address to use for the DB_HOST parameter.

Note: Other configuration issues can prevent the connection from being established. Verify that PostgreSQL is configured to accept connections from WSL 2. This includes its listen_addresses and authentication rules inside the pg_hba.conf file.

4. Verify the creation of the DAG using the Airflow standalone GUI.

The airflow standalone CLI command enables users to quickly spin up all core components of Airflow locally and simultaneously. This command also provides access to a web based graphic interface (GUI) that enables users to monitor the DAGs.

First, the admin credentials are required to access the GUI. To find the credentials: navigate inside the Airflow main installation package and look for either:

simple_auth_manager_passwords.json.generated

or

standalone_admin_password.txt

by running the following command:

cat simple_auth_manager_passwords.json.generated

This command returns the account credentials to access the web interface.

Figure 2 illustrates the commands used to run Airflow in standalone mode from WSL 2.

Figure 2. CLI commands to run Airflow in standalone mode from WSL 2.

I used the commands in Figure 2 to:

• Navigate to the Python application.
• Activate the Python virtual environment.
• Check which Airflow installation is being used.
• Run Airflow in standalone mode.

Once Airflow is running, navigate to localhost:8080. Airflow uses this port by default. Access to the web interface is granted after submitting user credentials (username and password).

Search for the DAG created for the pipeline by dag_id. In this case, the dag_id is etl_dag.

Figure 3 showcases the successful creation and registration of etl_dag.

Figure 3. Airflow web interface showcasing the successful creation and registration of the etl_dag.

Note: Make sure the DAG is activated by enabling the toggle.

5. Monitor the data loading in the PostgreSQL database.

To verify that the data is being loaded, access the PostgreSQL database using pgAdmin.

Figure 4 illustrates the insertion of the transformed data into the database using pgAdmin.

Figure 4. pgAdmin interface illustrating the successful insertion of the data into the weather_data table.

Conclusion

The orchestration was successfully implemented and verified. The DAG was created and registered, and its defined tasks were also validated. The transformed data was also successfully loaded into the PostgreSQL database.

It is important to note that the current implementation is not hosted. Therefore, Airflow must be manually started whenever the pipeline needs to be executed. To ensure continuous execution of the pipeline, it can be hosted on cloud infrastructure.

References

• Open-Meteo Weather API.
• GitHub repository of the project.
• Apache Airflow documentation.
• WSL documentation - Microsoft official website.

Trademark Notice

• Python and the Python logos are trademarks of the Python Software Foundation (PSF).
• Postgres, PostgreSQL and the Slonik Logo are trademarks or registered trademarks of the PostgreSQL Community Association of Canada, and used with their permission.
• Apache Airflow, Airflow, and the Airflow logo are trademarks of the Apache Software Foundation.
• Ubuntu is a registered trademark of Canonical Ltd.