ETL Pipelines with Python for AI Products

Key facts, with sources

• JetBrains' State of Python 2025 survey of more than 30,000 developers found that 51% of all Python developers are involved in data exploration and processing, with pandas and NumPy the most commonly used tools. (The JetBrains Blog)
• Apache Airflow reached more than 77,000 organizations using it as of November 2024, up from about 25,000 in 2020, with monthly downloads growing from 888,000 to over 31 million in the same period. (Astronomer State of Airflow 2025)
• The official Apache Airflow 2025 survey collected more than 5,250 responses from 116 countries, and event-driven scheduling introduced in Airflow 3 already showed almost 25% adoption. (Apache Airflow Blog)
• pandas has accumulated roughly 15.3 billion total downloads on PyPI, with around 700 million downloads in a single recent month. (pepy.tech)
• The Rust-based Polars DataFrame library passed 24 million monthly downloads and over 250 million total downloads as of September 2025, five years after its first commit. (Wikipedia)
• Gartner research estimates poor data quality costs organizations an average of at least $12.9 million per year. (Gartner)

Why AI products change the ETL playbook

Classic ETL feeds dashboards, where a bad row means a slightly wrong chart. In an AI product, transformed data becomes prompt context or embeddings, so errors are amplified: a mangled document gets chunked, embedded, retrieved, and confidently quoted back to a user. That raises the bar on three things — normalization quality, deduplication, and traceability from any model output back to the source record.

There's also a cost dimension that traditional ETL never had. Every record you load may trigger a paid embedding or enrichment call, so reprocessing unchanged data isn't just wasteful compute — it's a bill. I design AI-facing pipelines to be incremental from the first commit, not as a later optimization.

Design stages as pure, idempotent functions

I keep extract, transform, and load as separate Python functions with boring signatures: iterables in, iterators out. Generators mean the pipeline streams instead of holding the full dataset in memory, and each stage can be unit-tested with a list of three records.

Idempotency comes from deriving a stable ID for every record — usually a content hash — and loading with upsert semantics. If the pipeline crashes halfway and re-runs, nothing duplicates. This one decision eliminates the most common failure mode I see in code audits: pipelines that are only correct when they run exactly once, which no production pipeline ever does.

A minimal, production-shaped skeleton

Here is the shape I start nearly every project with. There's no framework — just generators, a content hash for identity, and batched upserts. The upsert uses the hash as the conflict key, so re-runs and overlapping backfills are safe by construction.

When a project outgrows this, the stages lift directly into Prefect tasks or Airflow operators because the business logic never knew about the orchestrator. That separation is deliberate: orchestrators should schedule and retry your functions, not own them.

import hashlib
import json
from collections.abc import Iterable, Iterator


def extract(paths: list[str]) -> Iterator[dict]:
    for path in paths:
        with open(path) as f:
            for line in f:
                yield json.loads(line)


def transform(records: Iterable[dict]) -> Iterator[dict]:
    for r in records:
        text = " ".join(r.get("body", "").split())
        if not text:
            continue
        yield {
            "id": hashlib.sha256(text.encode()).hexdigest(),
            "text": text,
            "source": r["source"],
        }


def load(records: Iterable[dict], batch_size: int = 500) -> None:
    batch: list[dict] = []
    for r in records:
        batch.append(r)
        if len(batch) >= batch_size:
            upsert_batch(batch)  # INSERT ... ON CONFLICT (id) DO NOTHING
            batch.clear()
    if batch:
        upsert_batch(batch)


load(transform(extract(input_files)))

Schema contracts between stages

Dictionaries flowing between stages work until the day a source system renames a field and your transform silently emits empty strings into the vector store. I put Pydantic models at every stage boundary: raw input gets validated on entry, and each stage emits a typed model the next stage can trust.

Validation failures don't crash the batch. They go to a quarantine table with the original payload and the error, the run continues, and the quarantine count becomes a monitored metric. A rising rejection rate is usually the first visible symptom of an upstream schema change — far better to catch it there than in retrieval quality complaints.

Loading for AI consumption: warehouse and vector store

AI products usually need a dual write: structured records into Postgres or a warehouse for filtering and analytics, and embedded chunks into a vector store for retrieval. I write the warehouse first and treat the vector store as a derived index — if the two ever disagree, the warehouse wins and the index gets rebuilt from it.

Embeddings get versioned by model identifier alongside the vector. When you upgrade embedding models — and you will — this turns a scary migration into a controlled re-embed: process rows where the stored model version doesn't match the current one, in batches, while the old index keeps serving.

Operational habits that keep it boring

Every run logs counts per stage: extracted, transformed, quarantined, loaded, skipped-as-unchanged. Those five numbers answer most incident questions before anyone opens a debugger, and a sudden drop in extracted counts catches dead upstream credentials the same day instead of weeks later.

Backfills are a first-class mode, not an emergency script — the same pipeline with a date-range parameter. Because loads are idempotent, a backfill overlapping normal runs is harmless. Retries with exponential backoff wrap only the I/O edges, never whole stages, so a flaky API doesn't force recomputing work that already succeeded.

When to hire senior help

Bring in a senior data engineer when pipeline failures start silently corrupting business metrics, or before committing to an orchestration and warehouse architecture, since storage layout and idempotency decisions are expensive to reverse once terabytes flow through them. A few weeks of experienced design work on schemas, retries, and backfill strategy routinely saves months of firefighting later. If your stack includes React Native + Python + AI, a senior engineer who owns the full product beats coordinating multiple juniors.

Bottom line

Dhairya Senjaliya ships Python — Data Processing Pipelines projects worldwide — book a scoping call to discuss your specific situation.

Common pitfalls to avoid

• ✕Running pipelines as cron jobs plus scripts with no idempotency or retry semantics, so a mid-run failure leaves half-written tables that corrupt downstream reports
• ✕Loading entire datasets into pandas in memory instead of chunking or using Polars or DuckDB, causing out-of-memory crashes the first time data volume grows 10x
• ✕Skipping schema and data-quality checks at ingestion, letting silent schema drift from a source API propagate wrong numbers into dashboards for weeks
• ✕Designing pipelines that cannot deterministically backfill historical data, making every bug fix or logic change a manual one-off reprocessing project

Frequently asked questions

Sources

• The JetBrains Blog
• Astronomer State of Airflow 2025
• Apache Airflow Blog
• pepy.tech
• Wikipedia
• Gartner