Build a Data Pipeline with Apache Airflow

Build a Data Pipeline with Apache Airflow: A Practical Guide

Imagine you’re manually downloading CSV files, cleaning them in Excel, and uploading them to a database every morning. It’s tedious, prone to errors, and impossible to scale. Now imagine clicking a single button to automate that entire workflow, with automatic retries, detailed logs, and a visual dashboard showing exactly where things succeed or fail. That’s the power of Apache Airflow, and it’s the industry standard for orchestrating data pipelines today.

If you’ve ever wished for a system that could manage your data workflows like a conductor leads an orchestra, you’re in the right place. In this guide, we’ll build a real, working data pipeline from scratch using Airflow. You’ll get code you can run today, not just theory.

Why Apache Airflow?

Airflow isn’t just another scheduler. It’s a platform designed to programmatically author, schedule, and monitor workflows. Its core philosophy is that workflows are defined as code (Python), making them versionable, testable, and reusable.

Unlike simpler tools like cron, Airflow gives you:

• Dynamic DAG generation: Create pipelines that adapt based on data.
• Visual dependency graphs: See exactly how tasks relate.
• Built-in retry logic: Automatically retry failed tasks.
• Rich UI: Monitor runs, view logs, and debug issues in real time [1].

Whether you’re building an ETL (Extract, Transform, Load) pipeline, automating ML training, or syncing databases, Airflow scales from a local script to a massive distributed cluster.

Step 1: Set Up Your Environment

Before writing code, let’s get Airflow running. The most reliable way is using Docker, as it handles all dependencies (PostgreSQL, Redis, web server) automatically.

Option A: Docker Compose (Recommended)

Create a folder called airflow-docker and download the official docker-compose.yaml:

mkdir airflow-docker
cd airflow-docker
curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.3.3/docker-compose.yaml'

Initialize the database and start services:

docker-compose up airflow-init
docker-compose up

After a few seconds, open http://localhost:8080 in your browser. Log in with the default credentials (airflow/airflow) from the official docs [1].

Option B: Local Installation (Quick Start)

If you don’t want Docker, install via pip:

pip install apache-airflow
airflow db init
airflow users create --role Admin --username admin --email [e-mail] --firstname Admin --lastname User --password admin

Create a dags folder in your AIRFLOW_HOME and place your pipeline files there [5].

Step 2: Define Your First DAG

A DAG (Directed Acyclic Graph) is the blueprint of your pipeline. It defines tasks and their execution order.

Let’s build a simple pipeline that:

1. Extracts weather data from a public API.
2. Transforms it (calculates averages).
3. Loads it into a SQLite database.

Create a file pipeline.py inside your dags folder:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import requests
import sqlite3

def extract_weather_data(**context):
    response = requests.get("https://api.open-meteo.com/v1/forecast?latitude=52.52&longitude=13.41&current=temperature_2m")
    data = response.json()
    context['ti'].xcom_push(key='weather_data', value=data)

def transform_weather_data(**context):
    ti = context['ti']
    raw_data = ti.xcom_pull(key='weather_data')
    temp = raw_data['current']['temperature_2m']
    transformed = {"temperature": temp, "timestamp": datetime.now().isoformat()}
    ti.xcom_push(key='transformed_data', value=transformed)

def load_to_db(**context):
    ti = context['ti']
    data = ti.xcom_pull(key='transformed_data')
    conn = sqlite3.connect("/usr/local/airflow/weather_db.sqlite")
    cursor = conn.cursor()
    cursor.execute("CREATE TABLE IF NOT EXISTS weather (temperature FLOAT, timestamp TEXT)")
    cursor.execute("INSERT INTO weather VALUES (?, ?)", (data['temperature'], data['timestamp']))
    conn.commit()
    conn.close()

with DAG(
    dag_id="weather_etl_pipeline",
    start_date=datetime(2024, 1, 1),
    schedule_interval="daily",
    catchup=False,
) as dag:

    extract_task = PythonOperator(
        task_id="extract",
        python_callable=extract_weather_data,
    )

    transform_task = PythonOperator(
        task_id="transform",
        python_callable=transform_weather_data,
    )

    load_task = PythonOperator(
        task_id="load",
        python_callable=load_to_db,
    )

    extract_task >> transform_task >> load_task

Key Concepts Explained

• PythonOperator: Runs Python functions as tasks.
• XComs: A mechanism to pass data between tasks (like xcom_push and xcom_pull) [3].
• Dependencies: The >> operator defines execution order: extract → transform → load [3].
• Idempotency: Each task should be safe to rerun without duplicating data (our CREATE TABLE IF NOT EXISTS ensures this) [1].

Step 3: Run and Monitor Your Pipeline

Once you save pipeline.py, the Airflow scheduler automatically detects it. Go to the UI:

1. Navigate to the DAGs page.
2. Find weather_etl_pipeline.
3. Click the toggle to enable it.
4. Click the play button (▶) to trigger a manual run [4].

You’ll see a graph view showing task progress. Green = success, red = failed. Click any task to view logs and debug errors [1].

Troubleshooting Tips

• If tasks fail, check the logs tab for Python errors.
• Ensure your dags folder path matches AIRFLOW_HOME/dags.
• For local installs, run airflow scheduler and airflow webserver in separate terminals [5].

Step 4: Make It Production-Ready

Your pipeline is working, but let’s add polish for real-world use.

Add Connection Management

Instead of hardcoding database paths, use Airflow’s Connections feature:

1. In the UI, go to Admin → Connections.
2. Click + to add a new connection.
3. Set Connection Type to Sqlite, Connection Id to weather_db_conn, and Host to your DB path [8].
4. Update your load_to_db function to use BaseHook.get_connection("weather_db_conn").

Set Up Alerts

Add email notifications for failures:

from airflow.operators.email import EmailOperator

on_failure_callback = EmailOperator(
    task_id="send_alert",
    to="[e-mail]",
    subject="DAG Failed: {{ dag.dag_id }}",
    html_content="Check the logs for details.",
)

Attach this to your DAG’s on_failure_callback parameter.

Schedule Automatically

Change schedule_interval="daily" to "@hourly" or a cron expression like "0 6 * * *" (runs at 6 AM daily) [6].

Why This Matters Now

Data pipelines are the backbone of modern analytics. Without orchestration, you’re stuck with manual scripts that break silently. Airflow gives you visibility, reliability, and scalability—all from a single Python file.

You don’t need a massive team to start. With Docker and this code, you can build a production-grade pipeline in under an hour.

Your Next Move

Don’t let this be just another tutorial you read and forget. Copy the code, run it, and break it. Then fix it. That’s how you learn.

• 🔧 Try modifying the pipeline: Add a task to send a Slack message after loading.
• 📚 Explore more: Check out Airflow’s official docs for operators like BashOperator, S3Operator, and PostgresOperator.
• 💬 Share your journey: Did you build something cool? Post it on Dev.to or Twitter and tag me!

The future of data is automated. Start building it today.

---

If you found this helpful, consider buying me a coffee ☕ — it keeps these articles coming!

Also check out my AI tools collection: AI 次元世界 — free AI tools for developers.

---

喜欢这篇文章？关注获取更多Python自动化内容！