Claude Code for Data Pipelines: ETL Development Workflow

Why ETL Work Is a Good Fit for Claude Code

Most data engineers didn't sign up to write boilerplate. You want to design good schemas, catch data quality problems before they hit production, and build pipelines that don't page you at 2 a.m. But the day-to-day reality of ETL work is full of repetitive, error-prone busywork: writing extraction connectors for the fifth API this quarter, hand-mapping column names between a source system and a warehouse, debugging a transformation that silently drops rows, or writing yet another set of assertions to check that a pipeline didn't just load a table full of nulls.

Claude Code changes the economics of that busywork. It's a terminal-native coding agent that can read your repository, run your pipeline scripts, inspect actual data samples, write and execute tests, and iterate against real error output — not just generate code from a prompt and hope it works. For ETL development specifically, this matters because so much of the job is investigative: you don't know the shape of the source data until you look at it, and you don't know if a transform is correct until you run it against a sample and check the output.

This article walks through a concrete workflow for using Claude Code across the ETL lifecycle: exploring unfamiliar schemas, scaffolding extraction and load code, writing and debugging transformation logic, building data quality checks, and wiring pipelines into orchestration tools. The goal isn't "let the AI write my pipeline" — it's using an agentic coding tool to compress the parts of pipeline development that are mechanical, so you spend your attention on the parts that require judgment: what data actually means, what "correct" looks like for your business, and where the pipeline is likely to break.

Setting Up Claude Code for a Data Project

Before touching any transformation logic, get Claude Code oriented in your project. If you're working in a monorepo with dbt models, Airflow DAGs, and a handful of Python extraction scripts, the agent needs to understand that layout before it can make useful changes.

Start a session in your pipeline repo and let Claude Code build context by reading the actual structure rather than guessing from folder names:

cd ~/projects/analytics-pipelines
claude

Once inside, ask it to survey the codebase before writing anything:

> Read through this repo and summarize the ETL architecture: what
  sources we extract from, what orchestrator runs the jobs, where
  transformation logic lives, and how data quality is currently
  checked. Don't write any code yet.

This single step matters more than it looks. Claude Code will open your dags/, models/, extractors/, and tests/ directories, read actual file contents, and give you a grounded summary instead of a generic description of "a typical ETL pipeline." If you have a CLAUDE.md file at the repo root, this is the place to document conventions once so you don't repeat them every session — things like "all timestamps are stored in UTC," "we never hard-delete rows, we soft-delete with a deleted_at column," or "staging tables are prefixed stg_ and marts are prefixed fct_/dim_." Claude Code reads this file automatically at the start of a session, so house rules about your data model become durable instructions instead of things you retype in every prompt.

Exploring Unfamiliar Source Data Before Writing Extraction Code

A common failure mode in ETL work is writing extraction code against an assumed schema, then discovering in production that a field is sometimes null, sometimes a string, sometimes missing entirely. Claude Code is useful here precisely because it can run code, not just write it — it can pull a sample from the source, inspect the actual shape of the data, and adjust the extraction logic based on what it finds instead of what the API docs claim.

Suppose you're building a new extractor for a third-party billing API. Ask Claude Code to investigate first:

> Hit the /invoices endpoint of the Stripe-like billing API described
  in docs/billing_api.md, pull 20 sample records, and show me the
  actual field types and any nulls or inconsistent structures you find.

Claude Code will write a small throwaway script, execute it, and read the output — then report back with something concrete, like "3 of 20 records have discount_amount as null rather than 0, and customer_id is sometimes an integer and sometimes a zero-padded string." That's the kind of detail that saves you a production incident. Only after this investigation does it make sense to write the actual extractor:

# extractors/billing_extractor.py
import requests
from datetime import datetime, timezone
from typing import Iterator

BASE_URL = "https://api.billingsystem.com/v1"

def fetch_invoices(api_key: str, since: datetime) -> Iterator[dict]:
    """Paginate through invoices modified since a given timestamp."""
    cursor = None
    headers = {"Authorization": f"Bearer {api_key}"}

    while True:
        params = {"updated_since": since.isoformat(), "limit": 100}
        if cursor:
            params["cursor"] = cursor

        resp = requests.get(f"{BASE_URL}/invoices", headers=headers, params=params)
        resp.raise_for_status()
        payload = resp.json()

        for record in payload["data"]:
            # Normalize inconsistent customer_id typing found during exploration
            record["customer_id"] = str(record["customer_id"]).zfill(8)
            # Coalesce null discount_amount to 0 rather than propagating null downstream
            record["discount_amount"] = record.get("discount_amount") or 0
            yield record

        cursor = payload.get("next_cursor")
        if not cursor:
            break

Notice that the normalization logic in this extractor isn't generic defensive coding — it's a direct response to the specific inconsistencies Claude Code found in the sample data. That's the difference between an agent that explores before it writes and one that pattern-matches a plausible-looking extractor from training data.

Writing Transformation Logic with Real Data in the Loop

Transformation code is where subtle bugs hide: a join that fans out rows unexpectedly, a timezone conversion applied twice, a GROUP BY that silently drops a category because of a case-sensitivity mismatch. The advantage of using Claude Code for this work is that it can run the transformation against a real (or representative sample) dataset and check the row counts, not just eyeball the SQL.

A workflow that catches these problems early:

1. Ask Claude Code to write the transform against a small local sample rather than the full warehouse table
2. Have it print row counts before and after each major step (join, filter, aggregation)
3. Ask it to flag any step where row count changes unexpectedly
4. Only then run it against the full dataset

> Write a dbt model that joins stg_orders to stg_customers on
  customer_id and aggregates total lifetime spend per customer.
  Before finalizing, run it against the seed data in
  seeds/sample_orders.csv and seeds/sample_customers.csv, print the
  row count of stg_orders, the row count after the join, and the row
  count after aggregation. Tell me if the join produces more rows
  than stg_orders had, since that would mean a fan-out.

The resulting model, with the kind of comment that documents a real decision rather than restating the code:

-- models/marts/fct_customer_lifetime_value.sql
with orders as (
    select
        customer_id,
        order_id,
        total_amount,
        order_status
    from {{ ref('stg_orders') }}
    where order_status != 'cancelled'  -- exclude cancelled orders from LTV
),

customers as (
    select customer_id, signup_date, region
    from {{ ref('stg_customers') }}
),

joined as (
    select
        c.customer_id,
        c.signup_date,
        c.region,
        o.order_id,
        o.total_amount
    from customers c
    -- left join preserves customers with zero orders, which matters
    -- for churn analysis downstream
    left join orders o on c.customer_id = o.customer_id
)

select
    customer_id,
    signup_date,
    region,
    count(distinct order_id) as total_orders,
    coalesce(sum(total_amount), 0) as lifetime_value
from joined
group by 1, 2, 3

Because Claude Code actually executed this against the seed data before presenting it, you get a concrete report alongside the code: "stg_orders has 1,240 rows, the join produces 1,240 rows (no fan-out), and aggregation collapses to 312 distinct customers, matching the customer count in stg_customers." That verification step is the difference between code that looks right and code that's been checked.

Debugging a Broken Pipeline from Actual Error Output

The most valuable use of Claude Code in day-to-day ETL work is debugging failures, because this is where an agent that can run commands and read logs outperforms a chat window where you paste in a stack trace and hope the model guesses right.

When a pipeline fails, resist the urge to describe the error from memory. Let Claude Code see it directly:

> Run `airflow tasks test daily_revenue_pipeline transform_orders 2026-07-02`
  and figure out why it's failing. Read the full traceback and check
  the upstream task's output before proposing a fix.

Claude Code will execute the task, capture the actual traceback, and — critically — go look at the upstream data rather than assuming the bug is in the transform itself. A representative example of what this catches: the transform step is raising a KeyError on discount_code, and the instinct is to add a .get() with a default. But if Claude Code checks the upstream extraction output first, it might find that the column was renamed in a recent source-system change and is now called promo_code — meaning the real fix is in the extractor's column mapping, not a defensive patch in the transform. That's a materially different fix, and you only find it by looking at the actual upstream output rather than guessing from the error message alone.

A pattern worth adopting deliberately: ask Claude Code to state its hypothesis before it changes any code.

> Before you fix anything, tell me what you think is causing this
  failure and how you'd verify it. Then verify it. Then propose the fix.

This forces a checkpoint where you can catch a wrong diagnosis before it turns into a wrong fix — important in pipelines where a "fix" that suppresses an error (like silently defaulting a missing field) can be worse than the original crash, because it hides a data quality problem instead of surfacing it.

Building Data Quality Checks as Part of the Workflow

ETL pipelines fail silently far more often than they fail loudly. A join that fans out, a currency field that's off by a factor of 100, a batch that loads zero rows because an upstream filter changed — none of these throw exceptions. They just produce wrong numbers that someone in finance or product notices three weeks later.

Claude Code is well suited to writing data quality checks because it can generate the check, run it against real data to confirm it actually fires on bad input, and iterate until the check is neither too strict (false alarms) nor too loose (misses real problems).

> Add data quality tests for fct_customer_lifetime_value: check that
  lifetime_value is never negative, that total_orders is never null,
  and that the row count doesn't drop by more than 5% day over day
  compared to the previous run. Use dbt's built-in test framework
  where possible, and write a custom test for the row count check.
  Then run the tests against yesterday's and today's data to confirm
  they actually pass on good data.

A custom dbt test for the row-count-drop check, generated and then verified against real snapshots:

-- tests/assert_no_significant_row_count_drop.sql
-- Fails if today's row count is more than 5% below yesterday's,
-- catching silent upstream filtering bugs or partial loads.

with today as (
    select count(*) as row_count
    from {{ ref('fct_customer_lifetime_value') }}
),

yesterday as (
    select row_count
    from {{ ref('row_count_history') }}
    where snapshot_date = current_date - interval '1 day'
)

select
    today.row_count as today_count,
    yesterday.row_count as yesterday_count
from today, yesterday
where today.row_count < yesterday.row_count * 0.95

The important habit here isn't the SQL itself — it's asking Claude Code to run the test against both a known-good dataset and a deliberately broken one (for example, a sample where you've manually dropped 10% of rows) before you trust it. A data quality check that's never been shown to fail is a check you haven't actually verified, and Claude Code can do that verification loop in a couple of minutes instead of you doing it by hand.

Refactoring and Scaling Pipeline Code

As pipelines mature, they accumulate duplicated extraction logic across sources, inconsistent naming, and transforms that were copy-pasted and lightly modified rather than properly parameterized. This is exactly the kind of mechanical, wide-reaching refactor where an agent that can read the whole repo and make coordinated changes earns its keep.

> We have five extractor files (extractors/stripe_extractor.py,
  extractors/hubspot_extractor.py, extractors/zendesk_extractor.py,
  extractors/shopify_extractor.py, extractors/mailchimp_extractor.py)
  that each implement their own retry logic and pagination handling.
  Extract the common pagination and retry logic into a shared base
  class in extractors/base.py, then refactor each extractor to use
  it. Run the existing extractor tests after each file to make sure
  nothing breaks.

A shared base class that comes out of this kind of refactor:

# extractors/base.py
import time
import requests
from typing import Iterator, Optional
from abc import ABC, abstractmethod

class PaginatedAPIExtractor(ABC):
    """Shared retry and pagination logic for REST-based extractors."""

    max_retries = 3
    backoff_seconds = 2

    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url

    @abstractmethod
    def build_params(self, cursor: Optional[str]) -> dict:
        ...

    @abstractmethod
    def parse_page(self, payload: dict) -> tuple[list[dict], Optional[str]]:
        """Return (records, next_cursor)."""
        ...

    def fetch_all(self, endpoint: str) -> Iterator[dict]:
        cursor = None
        while True:
            payload = self._get_with_retry(endpoint, self.build_params(cursor))
            records, cursor = self.parse_page(payload)
            yield from records
            if not cursor:
                break

    def _get_with_retry(self, endpoint: str, params: dict) -> dict:
        last_error = None
        for attempt in range(self.max_retries):
            try:
                resp = requests.get(
                    f"{self.base_url}/{endpoint}",
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    params=params,
                    timeout=30,
                )
                resp.raise_for_status()
                return resp.json()
            except requests.RequestException as e:
                last_error = e
                time.sleep(self.backoff_seconds * (attempt + 1))
        raise RuntimeError(f"Failed after {self.max_retries} retries") from last_error

The key discipline in a refactor like this is running the existing test suite after every single file change, not once at the end. If you have Claude Code touch all five extractors and only run tests afterward, a regression in the third file is indistinguishable from one in the fifth. Explicitly asking for per-file verification keeps the blast radius small and makes any failure easy to trace back to its cause.

Wiring Pipelines into Orchestration and CI

Once extraction, transformation, and quality checks are solid, the remaining work is operational: scheduling the pipeline, wiring it into CI so a bad pull request never reaches production, and making failures visible to the right people. Claude Code can scaffold this, but the actual DAG structure and alerting thresholds are judgment calls you should own, not delegate wholesale.

A useful pattern is to have Claude Code draft the DAG and then walk you through the dependency graph explicitly, rather than accepting it silently:

> Write an Airflow DAG for the daily revenue pipeline: extract
  from the billing API, load to the raw schema, run the dbt
  transform, then run the data quality tests. Explain the task
  dependencies you chose and why the quality tests run last, not
  in parallel with the load.

# dags/daily_revenue_pipeline.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="daily_revenue_pipeline",
    schedule_interval="0 6 * * *",
    start_date=datetime(2026, 1, 1),
    default_args=default_args,
    catchup=False,
) as dag:

    extract = PythonOperator(
        task_id="extract_billing_data",
        python_callable=run_billing_extraction,
    )

    load = PythonOperator(
        task_id="load_to_raw_schema",
        python_callable=load_raw_invoices,
    )

    transform = BashOperator(
        task_id="run_dbt_transform",
        bash_command="cd /opt/dbt && dbt run --select fct_customer_lifetime_value",
    )

    quality_check = BashOperator(
        task_id="run_dbt_tests",
        bash_command="cd /opt/dbt && dbt test --select fct_customer_lifetime_value",
    )

    extract >> load >> transform >> quality_check

Quality checks run strictly after the transform completes, not in parallel with it, because there's nothing to check until the transform has produced output — running them concurrently would either fail on a missing table or, worse, silently check stale data from the previous run. This is the kind of dependency reasoning worth confirming explicitly rather than trusting blindly, since a wrong ordering here fails silently rather than loudly.

For CI, a lightweight check that catches obviously broken pipeline code before merge:

> Add a GitHub Actions workflow that runs dbt compile and dbt test
  against a local Postgres instance with seed data whenever a pull
  request touches models/ or seeds/. Keep it fast — under 3 minutes.

Working Session Habits That Keep Output Trustworthy

A few habits separate ETL work where Claude Code is genuinely reliable from work where it quietly introduces problems:

• Always have it run before it reports. A transform that "should work" based on reading the SQL is not the same as one that's been executed against sample data with row counts printed. Ask for the run, not just the code.
• Ask for row counts at every stage, not just the final output. A pipeline that produces a plausible-looking final table can still have silently dropped 40% of rows in a middle step.
• Treat schema exploration as a separate step from code-writing. Don't let the agent write an extractor against an assumed schema — have it pull real samples first.
• Keep a `CLAUDE.md` with your data conventions. Column naming rules, timezone handling, soft-delete conventions, and which tables are considered source of truth all belong here so you're not re-explaining them every session.
• Use git branches for pipeline changes the same way you would for application code. Nothing about "it's just a SQL model" makes an untested change to a revenue table safe to run directly against production.
• Push back on defensive patches that hide root causes. If a fix is "default the missing field to zero," ask why the field is missing upstream before accepting that as the final answer.

None of these habits are unique to AI-assisted development — they're just good ETL discipline. What changes is that Claude Code makes it cheap enough to actually follow them every time, because the exploration, the verification run, and the row-count check no longer cost you twenty minutes of manual work per pipeline change.

Getting Started with Your Own Pipelines

The workflow above scales down as easily as it scales up. If you're maintaining a single nightly script that loads a CSV into a warehouse table, the same principles apply at a smaller size: let Claude Code inspect the actual input file before writing the loader, run the loader against a sample before the full file, and add one or two assertions that would catch a truncated or malformed load. You don't need an orchestrator or a dbt project to benefit from an agent that verifies its own work against real data.

If you're newer to agentic coding tools generally — not just for data work, but for the broader practice of directing an AI agent through a multi-step development task with verification along the way — that's exactly the skill set covered in our Claude Code Tutorial for Beginners course on teachyou.ai. It walks through setting up Claude Code, structuring CLAUDE.md context files, running multi-step agentic workflows, and building the habit of verifying output rather than trusting it blindly — all foundational skills that carry directly into ETL and data pipeline work like the examples in this article.