Automate Python Data Pipelines with Airflow

Automate Python Data Pipelines with Airflow: A Hands-On Guide

Let’s be real—manually running Python scripts to fetch, clean, and load data gets old fast. You set a cron job, hope it works, and panic when it fails at 3 AM. That’s where Apache Airflow comes in. It’s not just for big tech companies. PythonSkillset.com readers can use it to turn messy data workflows into reliable, scheduled pipelines.

I’ve seen teams waste hours debugging why a data pipeline broke two days ago. Airflow solves this by letting you define tasks as Python code, schedule them, and monitor everything from a web dashboard. No more guessing games.

Why Airflow Fits Python Developers

Airflow is Python-native. Your existing scripts—whether they use Pandas, Requests, or SQLAlchemy—can become tasks in a DAG (Directed Acyclic Graph). DAG is just a fancy term for “a sequence of steps that must run in order.”

For example, imagine you need to:

1. Scrape weather data from an API
2. Clean missing values
3. Load it into a PostgreSQL database

In Airflow, you write each step as a Python function wrapped in a task. Airflow handles retries, logging, and dependencies. If step 2 fails, Airflow knows not to run step 3 until you fix it.

Setting Up Your First Pipeline

Let’s build a simple pipeline together. Assume you have a PythonSkillset.com project that collects daily cryptocurrency prices.

Step 1: Install Airflow

pip install apache-airflow

Then initialize the database:

airflow db init

Step 2: Write Your DAG

Create a file called crypto_pipeline.py in the dags/ folder:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests
import pandas as pd
import sqlalchemy

def fetch_prices():
    response = requests.get("https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd")
    data = response.json()
    return data['bitcoin']['usd']

def clean_data(**context):
    price = context['task_instance'].xcom_pull(task_ids='fetch_prices')
    # Simple validation
    if price <= 0:
        raise ValueError("Invalid price!")
    return price

def load_to_db(**context):
    price = context['task_instance'].xcom_pull(task_ids='clean_data')
    engine = sqlalchemy.create_engine('postgresql://user:pass@localhost/mydb')
    df = pd.DataFrame({'price': [price], 'timestamp': [datetime.now()]})
    df.to_sql('crypto_prices', engine, if_exists='append', index=False)

default_args = {
    'owner': 'pythonskillset',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'crypto_pipeline',
    default_args=default_args,
    description='Fetch crypto prices every hour',
    schedule_interval='0 * * * *',  # Every hour
    start_date=datetime(2025, 1, 1),
    catchup=False
)

with dag:
    fetch = PythonOperator(task_id='fetch_prices', python_callable=fetch_prices)
    clean = PythonOperator(task_id='clean_data', python_callable=clean_data)
    load = PythonOperator(task_id='load_to_db', python_callable=load_to_db)

    fetch >> clean >> load  # Defines order

Step 3: Start and Monitor

airflow standalone

Visit http://localhost:8080 (login with admin/admin). You’ll see your DAG. Toggle it on, and Airflow runs it every hour. If a task fails, you get error logs right in the UI.

Real-World Lessons

A PythonSkillset.com client of mine runs a daily revenue report. Before Airflow, they manually ran a Jupyter notebook. One day, the notebook crashed due to a bad API key. They didn’t notice for three days. Now, Airflow alerts them via email the instant something breaks.

Another scenario: You have three data sources—CSV, MySQL, and an API. Airflow can run them in parallel, then merge results. This cuts execution time from 40 minutes to 12.

Common Pitfalls to Avoid

• Hardcoding credentials: Use Airflow’s Connections or environment variables. Never put passwords in DAG files.
• Ignoring catchup: For historical data, set catchup=True to backfill. For real-time, keep it False.
• Too many dependencies: Keep DAGs simple. If a DAG has 50 tasks, consider splitting it.

Taking It Further

Once you’re comfortable, explore Airflow’s sensors (wait for a file to arrive) or the TaskFlow API (simpler syntax). For teams, the RBAC security model lets you control who can edit pipelines.

Airflow does have a learning curve—especially for complex DAGs. But for most Python developers, it’s the missing piece between “scripts that run” and “pipelines you can trust.”

Start with a small DAG today. Your future self—and your data team—will thank you.