Project Overview

In this project, you’ll build a daily ETL (Extract, Transform, Load) pipeline using Apache Airflow. The pipeline will extract data from a public API, transform it into a clean format, and load it into a database. This project simulates a common real-world data engineering workflow.

Learning Objectives

By the end of this project, you will be able to:

  • Define a multi-task DAG in Airflow.

  • Extract data from an external API.

  • Transform raw JSON data into a structured format.

  • Load the transformed data into a database table.

  • Schedule the pipeline to run daily.

Requirements

  • Apache Airflow installed (Docker or local setup).

  • Python 3.8+

  • PostgreSQL (or any relational database) for storage.

  • API: REST Countries API (provides country data in JSON format).

Project Architecture

Pipeline Steps (DAG):

  1. Extract → Fetch country data from API.

  2. Transform → Clean and normalize JSON into tabular format.

  3. Load → Insert transformed data into a database table.

Visually:

 
Extract → Transform → Load

DAG Implementation

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
import requests
import pandas as pd
from datetime import datetime

# Step 1: Extract
def extract_data(**context):
    url = "https://restcountries.com/v3.1/all"
    response = requests.get(url)
    data = response.json()
    context['ti'].xcom_push(key='raw_data', value=data)

# Step 2: Transform
def transform_data(**context):
    data = context['ti'].xcom_pull(key='raw_data', task_ids='extract_task')
    df = pd.json_normalize(data)
    df_clean = df[['name.common', 'region', 'population']].rename(
        columns={'name.common': 'country', 'region': 'region', 'population': 'population'}
    )
    context['ti'].xcom_push(key='clean_data', value=df_clean.to_dict('records'))

# Step 3: Load
def load_data(**context):
    records = context['ti'].xcom_pull(key='clean_data', task_ids='transform_task')
    df = pd.DataFrame(records)

    hook = PostgresHook(postgres_conn_id="my_postgres")
    engine = hook.get_sqlalchemy_engine()
    df.to_sql("countries", engine, if_exists="replace", index=False)

# Define DAG
with DAG(
    dag_id="daily_etl_pipeline",
    description="ETL pipeline with Extract-Transform-Load steps",
    start_date=datetime(2025, 1, 1),
    schedule_interval="@daily",
    catchup=False
) as dag:

    extract_task = PythonOperator(
        task_id="extract_task",
        python_callable=extract_data,
        provide_context=True
    )

    transform_task = PythonOperator(
        task_id="transform_task",
        python_callable=transform_data,
        provide_context=True
    )

    load_task = PythonOperator(
        task_id="load_task",
        python_callable=load_data,
        provide_context=True
    )

    extract_task >> transform_task >> load_task

Key Notes

  • Extract → Uses requests to fetch data from API.

  • Transform → Uses pandas to clean and reshape JSON data.

  • Load → Uses PostgresHook to insert data into a database table.

  • XComs → Used to pass data between tasks.

Expected Output

  • A PostgreSQL table countries containing:

countryregionpopulation
FranceEurope67081000
JapanAsia125960000
BrazilAmericas213993000

  • The pipeline runs daily at midnight, refreshing the table.

Project Extensions (Optional)

  • Add error handling & retries in case API fails.

  • Store raw data in S3/GCS before transformation.

  • Send a Slack notification after successful load.

  • Add a Branching Operator to skip load if extract fails.

Deliverables

  1. daily_etl_pipeline.py DAG file in your Airflow dags/ folder.

  2. A countries table in PostgreSQL updated daily.

  3. Screenshots of Graph View, Logs, and Data Output.

PreviousNext