Data Pipelines for AI: Feeding the Beast at Scale

Every AI system is only as good as its data. You can have the most sophisticated model architecture, the best hardware, infinite compute — but if your data pipeline is garbage, your model is garbage.

Garbage in, garbage out isn't just a cliché in AI. It's the #1 reason production models fail.

Think about it: ChatGPT was trained on trillions of tokens from the internet. Someone had to crawl it, clean it, deduplicate it, filter the toxic content, format it into training-ready batches, and stream it to 50,000 GPUs. That's not a script. That's a multi-stage, fault-tolerant, distributed data pipeline. And it's way harder than the model training itself.

The AI Data Pipeline: End to End

Every AI data pipeline has the same five stages, whether you're building a chatbot or training a foundation model:

┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐

│ Ingest │───▶│ Clean │───▶│Transform │───▶│ Store │───▶│ Serve │

│ │ │ │ │ │ │ │ │ │

│ APIs │ │ Dedup │ │ Chunk │ │ S3/GCS │ │ Training │

│ Scrapers │ │ Filter │ │ Tokenize │ │ Vector DB│ │ Feature │

│ DBs │ │ Validate │ │ Embed │ │ Data Lake│ │ Store │

│ Streams │ │ Fix │ │ Enrich │ │ │ │ Inference│

└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘

Let's walk through each stage.

Stage 1: Ingestion — Getting the Data

Data comes from everywhere, and it's never in the format you want.

Batch Ingestion

python

Pull data from multiple sources on a schedule

import pandas as pd

from sqlalchemy import create_engine

From a database

engine = create_engine("postgresql://localhost/production")
orders = pd.read_sql("SELECT * FROM orders WHERE date > '2024-01-01'", engine)

From an API

import requests

response = requests.get("https://api.internal/customer-interactions",
params={"since": "2024-01-01"})
interactions = pd.DataFrame(response.json())

From file storage

documents = []

for file in s3.list_objects("s3://raw-data/documents/"):

content = s3.get_object(file)
    documents.append(parse_document(content))

Stream Ingestion

python

Real-time data from Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
'user-events',
    bootstrap_servers=['kafka:9092'],
    auto_offset_reset='earliest'

)

for message in consumer:

event = json.loads(message.value)
    process_event(event)  # → cleaning pipeline

The key decision: Batch or stream?

Factor Batch Stream

Freshness Hours to daily Seconds to minutes

Complexity Lower Higher

Cost Cheaper More expensive

Rule of thumb: Start with batch. Add streaming only for features that must be fresh. 90% of AI data pipelines are batch.

🔧

Stage 2: Cleaning — The Unglamorous Truth

Data cleaning is 80% of the work and 0% of the glory. But it's where model quality is actually determined.

For Text Data (LLM Training / RAG)

python

import re

from langdetect import detect

def clean_text(raw_text):

# Remove HTML artifacts
    text = re.sub(r'<[^>]+>', '', raw_text)

# Remove excessive whitespace
    text = re.sub(r'\s+', ' ', text).strip()

# Remove boilerplate (navigation, headers, footers)
    text = remove_boilerplate(text)

# Language filter (keep only English, or your target language)
    try:
        if detect(text) != 'en':
            return None  # Skip non-English
    except:
        return None  # Can't detect = probably garbage

# Quality filter
    if len(text) < 100:
        return None  # Too short to be useful
    if len(text) > 100000:
        return None  # Probably a data dump, not real content

# Deduplication check
    text_hash = hashlib.md5(text.encode()).hexdigest()
    if text_hash in seen_hashes:
        return None  # Duplicate
    seen_hashes.add(text_hash)

return text

For Structured Data (Traditional ML)

python

def clean_structured(df):

# Handle missing values
    df['age'] = df['age'].fillna(df['age'].median())
    df['category'] = df['category'].fillna('unknown')

# Remove outliers
    df = df[df['amount'] < df['amount'].quantile(0.99)]

# Fix data types
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['user_id'] = df['user_id'].astype(str)

# Validate ranges
    df = df[df['age'].between(0, 120)]
    df = df[df['amount'] > 0]

return df

Deduplication at Scale

Exact deduplication (hash matching) is easy. Near-deduplication is the real challenge — finding documents that are 95% identical but not exactly the same (same article on different websites, slightly reformatted content).

python

MinHash for near-deduplication at scale

from datasketch import MinHash, MinHashLSH

lsh = MinHashLSH(threshold=0.8, num_perm=128)

for doc_id, text in documents:

minhash = MinHash(num_perm=128)
    for word in text.split():
        minhash.update(word.encode('utf8'))

# Check if similar document already exists
    result = lsh.query(minhash)
    if not result:
        lsh.insert(doc_id, minhash)
        # Keep this document
    else:
        # Near-duplicate found, skip it
        pass

The RedPajama and Dolma datasets use MinHash + exact deduplication. For a 1-trillion-token dataset, dedup can remove 30-40% of the data. That's not waste — it's preventing your model from memorizing repeated content.

📊

Stage 3: Chunking — Breaking Data for AI Consumption

For RAG and embedding pipelines, you need to split documents into chunks that are small enough to embed but large enough to be meaningful.

python

Naive chunking (don't do this)

chunks = [text[i:i+500] for i in range(0, len(text), 500)]

❌ Splits mid-sentence, mid-word, mid-thought

Better: Recursive character splitting

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]

)

chunks = splitter.split_text(text)

✓ Respects paragraph boundaries, then sentence boundaries

Best: Semantic chunking

from langchain_experimental.text_splitter import SemanticChunker

from langchain_openai import OpenAIEmbeddings

chunker = SemanticChunker(
OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"

)

chunks = chunker.split_text(text)

✓ Splits where the MEANING changes, not where characters end

Strategy Speed Quality Cost

Fixed-size ⚡ Fast ❌ Poor Free

Apache Airflow: Orchestrating the Pipeline

Individual scripts are fine for prototypes. Production needs orchestration — scheduling, dependencies, retries, monitoring, alerting.

Apache Airflow is the industry standard for batch data pipelines.

python

dags/ai_data_pipeline.py

from airflow import DAG

from airflow.operators.python import PythonOperator

from datetime import datetime, timedelta

default_args = {
'retries': 3,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,

}

dag = DAG(
'ai_training_data_pipeline',
    default_args=default_args,
    schedule_interval='@daily',
    start_date=datetime(2024, 1, 1),
    catchup=False,

)

ingest = PythonOperator(
task_id='ingest_raw_data',
    python_callable=ingest_from_sources,
    dag=dag,

)

clean = PythonOperator(
task_id='clean_and_validate',
    python_callable=clean_data,
    dag=dag,

)

chunk = PythonOperator(
task_id='chunk_documents',
    python_callable=chunk_and_embed,
    dag=dag,

)

store = PythonOperator(
task_id='store_to_vector_db',
    python_callable=upsert_to_qdrant,
    dag=dag,

)

compute_features = PythonOperator(
task_id='compute_features',
    python_callable=compute_and_push_features,
    dag=dag,

)

Define dependencies

ingest >> clean >> [chunk, compute_features]

chunk >> store

The Airflow UI shows you a visual DAG of your pipeline, with green/red/yellow indicators for each stage. When something fails at 3 AM, you know exactly which stage broke, you can see the logs, and you can retry from that point — not from scratch.

Alternatives to Airflow:

Tool Best For Complexity

Airflow Complex batch pipelines High

Prefect Python-native, modern Airflow Medium

Dagster Data-asset focused Medium

Streaming Pipelines: When Batch Isn't Fast Enough

For real-time features and live data processing:

python

Apache Flink (via PyFlink) for stream processing

from pyflink.datastream import StreamExecutionEnvironment

from pyflink.table import StreamTableEnvironment

env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

Read from Kafka stream

t_env.execute_sql("""

CREATE TABLE user_events (
        user_id STRING,
        event_type STRING,
        amount DOUBLE,
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )

""")

Compute streaming features

t_env.execute_sql("""

INSERT INTO feature_store
    SELECT
        user_id,
        COUNT(*) OVER w AS transaction_count_1h,
        AVG(amount) OVER w AS avg_amount_1h,
        MAX(amount) OVER w AS max_amount_1h
    FROM user_events
    WINDOW w AS (
        PARTITION BY user_id
        ORDER BY event_time
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
    )

""")

🔬

Data Quality: The Pipeline Within the Pipeline

Every AI data pipeline needs a quality gate:

python

Great Expectations for data validation

import great_expectations as gx

context = gx.get_context()

Define expectations

validator = context.sources.pandas_default.read_dataframe(df)

validator.expect_column_values_to_not_be_null("user_id")

validator.expect_column_values_to_be_between("amount", 0, 100000)

validator.expect_column_values_to_be_in_set("category",

["electronics", "clothing", "food", "other"])

validator.expect_table_row_count_to_be_between(10000, 10000000)

Run validation

results = validator.validate()

if not results.success:

raise DataQualityError(f"Pipeline failed quality checks: {results}")

Track data quality metrics over time:

Row count (sudden drops = ingestion failure)

Null rates (increasing = source degradation)

Distribution shifts (changing = data drift)

Schema changes (columns added/removed = breaking change)

🛡️

Practical Takeaways

Data pipelines are 80% of AI engineering — the model is the easy part

Start with batch, add streaming only when needed — complexity has a cost

Deduplication is non-negotiable — duplicates degrade model quality silently

Use semantic chunking for RAG — fixed-size chunks lose context at boundaries

Orchestrate with Airflow or Prefect — cron jobs don't scale, don't retry, don't alert

Quality gates at every stage — catch bad data before it poisons your model

📦

What's Next?

Episode 89: Scaling Laws — You've got data, models, compute. But how much of each do you actually need? Scaling laws tell you the mathematically optimal ratio of data to parameters to compute. Chinchilla changed everything — and the answer to "should I train a bigger model?" is almost always "no."

← Previous

Ep 87: Feature Stores

Next →

Ep 89: Scaling Laws

Next: Episode 89 — Scaling Laws

OpenAI spent $1 billion training GPT-5. Google poured billions into Gemini Ultra. Are scaling laws hitting a wall?

This is part of a 98-episode series covering AI engineering from tokens to production deployment.