Building Your First Airflow DAG for ETL: A Practical Guide

If you are getting started with Airflow and want to build your first DAG to move some data around, this article walks you through exactly that. We will build a simple ETL pipeline that extracts data from an API, transforms it, and loads it into a Postgres table. Nothing fancy — just enough to understand how DAGs, tasks, and scheduling hang together.

I have seen people jump straight into the Airflow UI and get overwhelmed by the number of options. The best way to learn Airflow is to write a DAG, run it, break it, then fix it. That is what we will do here.

What We Are Building

We will pull a list of GitHub repositories for a user via the GitHub API, pick out the fields we care about (repo name, stars, language, last updated), and insert those rows into a local Postgres table. The DAG will run once a day.

The point is not the specific source or destination — you can swap in any API and any database. The point is understanding the structure of a DAG and how Airflow executes it.

Prerequisites

I am assuming you have Airflow running locally. If you do not, the quickest way is with the official Docker Compose file:

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.9.0/docker-compose.yaml'
mkdir -p ./dags ./logs ./plugins
docker compose up -d

You will also need a Postgres instance you can write to. If you are running Airflow via Docker, you can spin one up:

docker run -d --name postgres-etl -p 5433:5432 \
  -e POSTGRES_PASSWORD=testpass \
  -e POSTGRES_DB=github_repos \
  postgres:15

We are using port 5433 to avoid clashing with Airflow’s own Postgres metadata database.

Step 1: The DAG File

Create a file called github_repo_etl.py inside your dags/ folder. This is where Airflow looks for DAG definitions by default.

Start with the imports and the default arguments:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
import requests
import logging

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': False,
}

A couple of things to point out here. start_date tells Airflow when this DAG first becomes active. It does not mean the DAG starts running from that exact moment — it gives Airflow a reference point for scheduling. depends_on_past: False means each run is independent. If yesterday’s run failed, today’s run still kicks off normally. For an ETL pipeline, that is usually what you want.

Step 2: Define the DAG

dag = DAG(
    'github_repo_etl',
    default_args=default_args,
    description='Pull GitHub repos for a user and load into Postgres',
    schedule_interval='@daily',
    catchup=False,
    tags=['github', 'api', 'postgres'],
)

Two settings worth understanding:

• schedule_interval='@daily' means this DAG runs at midnight every day. If you need a specific time, use a cron expression like '0 6 * * *' for 6 AM.
• catchup=False stops Airflow from backfilling all the days between start_date and today. If you are building a daily ETL, you probably do not want hundreds of runs queued the first time you deploy. Turn it on only when you need historical data replayed.

Step 3: Extract — Pull Data from GitHub

def extract_github_repos(**kwargs):
    username = 'apache'  # swap with your target user or org
    url = f'https://api.github.com/users/{username}/repos?per_page=100'
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    repos = response.json()

    logging.info(f'Fetched {len(repos)} repos for user {username}')

    # Push to XCom so the next task can use it
    kwargs['ti'].xcom_push(key='repos', value=repos)

The PythonOperator lets you run any Python callable as a task. We use XCom (cross-communication) to pass data between tasks. XCom stores the data in Airflow’s metadata database, so keep the payload reasonable — do not shove megabytes of data through it. For large datasets, write to a file on shared storage or object storage like S3/GCS instead.

Step 4: Transform — Pick the Fields We Need

def transform_repo_data(**kwargs):
    ti = kwargs['ti']
    repos = ti.xcom_pull(key='repos', task_ids='extract_github_repos')

    rows = []
    for repo in repos:
        rows.append({
            'name': repo.get('name'),
            'stars': repo.get('stargazers_count', 0),
            'language': repo.get('language') or 'Unknown',
            'last_updated': repo.get('updated_at'),
            'description': (repo.get('description') or '')[:200],
        })

    logging.info(f'Transformed {len(rows)} rows')
    ti.xcom_push(key='transformed_rows', value=rows)

Nothing clever here — just trimming the API response down to what we actually want. Truncating the description at 200 characters avoids surprise column-width issues later.

Step 5: Load — Insert into Postgres

def load_to_postgres(**kwargs):
    ti = kwargs['ti']
    rows = ti.xcom_pull(key='transformed_rows', task_ids='transform_repo_data')

    pg_hook = PostgresHook(postgres_conn_id='my_postgres')
    conn = pg_hook.get_conn()
    cursor = conn.cursor()

    create_sql = '''
        CREATE TABLE IF NOT EXISTS github_repos (
            name TEXT,
            stars INTEGER,
            language TEXT,
            last_updated TIMESTAMP,
            description TEXT,
            ingested_at TIMESTAMP DEFAULT NOW()
        )
    '''
    cursor.execute(create_sql)

    insert_sql = '''
        INSERT INTO github_repos (name, stars, language, last_updated, description)
        VALUES (%s, %s, %s, %s, %s)
        ON CONFLICT (name) DO UPDATE SET
            stars = EXCLUDED.stars,
            language = EXCLUDED.language,
            last_updated = EXCLUDED.last_updated,
            description = EXCLUDED.description
    '''

    for row in rows:
        cursor.execute(insert_sql, (
            row['name'], row['stars'], row['language'],
            row['last_updated'], row['description']
        ))

    conn.commit()
    cursor.close()
    conn.close()
    logging.info(f'Loaded {len(rows)} rows into Postgres')

A few things worth noting:

• We use ON CONFLICT ... DO UPDATE so we can run this DAG daily without duplicating rows. If a repo already exists, we update the stats instead of inserting a duplicate. You need a unique constraint or primary key on the name column for this to work — add it if your table does not already have one.
• The PostgresHook uses a connection ID (my_postgres) that you configure in the Airflow UI under Admin → Connections. This keeps credentials out of the DAG code.
• We close the cursor and connection explicitly. Airflow’s hooks usually handle this, but being explicit in custom Python code avoids connection leaks when things go wrong.

Step 6: Wire It All Together

extract_task = PythonOperator(
    task_id='extract_github_repos',
    python_callable=extract_github_repos,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_repo_data',
    python_callable=transform_repo_data,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load_to_postgres',
    python_callable=load_to_postgres,
    dag=dag,
)

extract_task >> transform_task >> load_task

The >> operator sets the dependencies: extract runs first, then transform, then load. If extract fails, transform and load do not run. This is the simplest pattern but exactly what you want for a pipeline where each step depends on the previous one.

Running and Checking the DAG

Once the DAG file is in your dags/ folder, Airflow picks it up within a minute or two (the default dag_dir_list_interval is 30 seconds). Go to the Airflow UI, find github_repo_etl, and toggle it on. Then click the play button to trigger a manual run.

You can watch the task statuses change in the Graph or Gantt views — the Gantt view is especially useful when you want to see how long each task takes.

Airflow Connections You Need

Before the DAG runs successfully, you need to set up the Postgres connection. Go to Admin → Connections in the Airflow UI and add:

Field	Value
Connection ID	my_postgres
Connection Type	Postgres
Host	postgres (or your hostname)
Schema	github_repos
Login	postgres
Password	testpass
Port	5432

If you are running Postgres outside Docker Compose, use host.docker.internal or your machine’s IP instead of localhost, because from inside the Airflow container, localhost points to the container itself, not your host.

Comparison: PythonOperator vs Other Operators

Airflow has a bunch of operators. Here is when I reach for each one:

Operator	Best For	Watch Out For
PythonOperator	Custom logic, API calls, anything not covered by a built-in	XCom for large data is slow; write to storage instead
BashOperator	Quick scripts, calling CLI tools	Hard to test, no structured error handling
PostgresOperator / SQLExecuteQueryOperator	Running SQL directly	Less flexible than Python if you need conditional logic
S3ToPostgresOperator	Straight data movement without transforms	Only works if you do not need to reshape the data
TaskFlow @task decorator	Cleaner syntax for passing data between tasks	Still uses XCom under the hood, same size limits

For our ETL, the PythonOperator made sense because we had custom transformation logic between extract and load. If your pipeline is purely moving files from S3 to Redshift with no transforms, use the built-in transfer operators — less code, less to break.

Production Considerations

This DAG works fine locally, but here is what you would change before deploying to a real environment:

1. Use a dedicated secrets backend. Hardcoding API usernames or relying on the Airflow connections UI is not great at scale. Use HashiCorp Vault, AWS Secrets Manager, or Airflow’s built-in secrets backend.

2. Write to object storage, not XCom. XCom is fine for small metadata (row counts, filenames). For the actual data, each task should write to a GCS or S3 bucket. The next task reads from there. This makes it easier to debug — you can inspect the intermediate files directly — and avoids bloating Airflow’s database.

3. Add a failure notification. Right now we have email_on_failure: False. In production, you want an on_failure_callback that posts to Slack or sends an email. Nobody checks the Airflow UI every morning.

4. Handle API rate limits. The GitHub API allows 60 unauthenticated requests per hour. Add an API token and a time.sleep() or exponential backoff if you hit a 429. For high-volume pipelines, use a dedicated data source (a database replica or an export) rather than hammering a public API.

5. Make the DAG idempotent. Our ON CONFLICT upsert helps, but if the API returns different results between runs (deleted repos, renamed repos), you need logic to handle that. A common pattern is to add a deleted_at column and soft-delete repos that disappear from the API instead of silently keeping stale data.

6. Use a sensor if the source is unreliable. If your API has known downtime windows, add an HttpSensor or a custom sensor task before the extract step to wait until the endpoint is healthy before running the pipeline.

Things That Trip Beginners Up

• start_date is not the current time. If you set start_date=datetime(2026, 1, 1) and schedule_interval='@daily', the first run is for January 2nd at midnight — because Airflow runs a schedule interval after the period ends. There is a whole logic around this called the execution date. Just remember: the run for 2026-01-27 actually fires after 2026-01-27 ends.

• DAGs are parsed frequently. Airflow parses every DAG file every 30 seconds (by default). Do not put expensive computations or network calls at the module level. All the imports and the DAG definition run every time the scheduler parses the file. Keep the DAG file lightweight — put heavy logic inside the callables.

• XCom keys are case-sensitive. xcom_pull(key='Repos') will not find what xcom_push(key='repos') stored. Use consistent, lowercase keys to avoid silent failures.

Wrapping Up

We built a simple Airflow DAG that pulls data from an API, transforms it, and loads it into Postgres. The DAG structure — a few PythonOperators wired together with >> — covers a surprising number of real-world ETL use cases.

If you are new to Airflow, the best thing you can do is build something like this from scratch, deploy it, and watch it run for a few days. You will run into scheduling quirks, connection misconfigurations, and surprises with XCom — and you will learn more from fixing those than from any tutorial.

Once this is solid, the next step is replacing PythonOperator with the TaskFlow @task decorator for cleaner code, and moving intermediate data to a proper storage layer instead of XCom. But that is a topic for another post.