RAG Pipelines for Data Engineers: You Already Know How to Build This

Strip away the hype and a RAG (retrieval-augmented generation) system is two pipelines — one batch, one online. If you can build an ETL pipeline, you can build RAG. The model is the cheap part. Freshness, chunking, access control, and evals are the job.

flowchart LR
  docs[Documents] --> hash[Hash + incremental]
  hash --> chunk[Chunk + metadata]
  chunk --> embed[Embed]
  embed --> store[Vector store]
  q[User question] --> retrieve[Retrieve k + filters]
  store --> retrieve
  retrieve --> llm[LLM generate + citations]
  llm --> answer[Grounded answer]

Batch ingest on the left. Online retrieve-then-generate on the right. Same index.

The architecture in data engineering terms

Ingestion pipeline (batch — this is 80% of the work):

Extract documents (wikis, PDFs, tickets, database rows)
Transform: clean → split into chunks → embed each chunk into a vector
Load chunks + vectors + metadata into a vector store

Query pipeline (online):

Embed the user’s question with the same model
Retrieve the top-k most similar chunks, with metadata filters
Stuff them into the LLM prompt and generate an answer with citations

Sound familiar? It’s ELT with an unusual index at the end. Warehouse-native variants of the same shape live in the Cortex guide. Store choice is a separate decision — see the vector database comparison.

Extract: treat documents like a source system

List every source as you would a warehouse source. For each one, write down:

Field	Why it matters
`source_system`	Filter at query time (“finance wiki only”)
`doc_id`	Idempotent upserts and deletes
`acl` / `group`	Retrieval must not leak across teams
`updated_at`	Incremental window
`content_hash`	Skip unchanged docs
`embed_model`	Rebuild when the model changes

PDFs and HTML need a clean step before chunking. Tables that become a blob of words retrieve as garbage. Prefer a structured extract (heading, table rows as text, page number) over dumping raw bytes into the embedder.

Deletes are part of extract. If a page leaves the wiki, its chunks must leave the index. A weekly “source of truth minus index” reconciliation job is not optional once you have more than one writer.

Transform: chunking is the business logic

Fixed character windows are a prototype. Production chunking follows document structure:

Split on headings, then paragraphs, then a size cap
Keep a small overlap so a sentence that straddles a boundary still retrieves
Never split a markdown table mid-row if you can avoid it
Attach the heading path to every chunk (Finance / Q3 close / Accruals)

A heading-aware splitter beats a clever embedding model on policy docs.

import re

def chunk_by_heading(text: str, size: int = 800, overlap: int = 150) -> list[dict]:
    parts = re.split(r"(?m)^#{1,3} ", text)
    headings = re.findall(r"(?m)^#{1,3} (.+)$", text)
    out: list[dict] = []
    for i, part in enumerate(parts):
        heading = headings[i - 1] if i and i - 1 < len(headings) else ""
        body = part.strip()
        if not body:
            continue
        step = max(size - overlap, 1)
        for j in range(0, len(body), step):
            piece = body[j : j + size]
            if piece.strip():
                out.append({"heading": heading, "text": piece})
    return out

Use the character splitter only as a fallback when the source has no structure (chat logs, raw tickets). Overlap of 10–20% is enough. Huge overlap just duplicates tokens and cost.

Load: embeddings are a derived table

Pin the embedding model. Store it on every row. If you change text-embedding-3-small to a different dimension, the old index is invalid — rebuild, do not mix.

# pip install openai chromadb
import hashlib
import chromadb
from openai import OpenAI

EMBED_MODEL = "text-embedding-3-small"
llm = OpenAI()
store = chromadb.PersistentClient("./rag_db").get_or_create_collection("docs")


def content_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def ingest(doc_id: str, text: str, source: str, acl: str) -> int:
    digest = content_hash(text)
    existing = store.get(where={"doc_id": doc_id}, include=["metadatas"])
    if existing["ids"] and existing["metadatas"][0].get("content_hash") == digest:
        return 0  # unchanged — skip the embed call

    if existing["ids"]:
        store.delete(ids=existing["ids"])

    pieces = chunk_by_heading(text)
    chunks = [p["text"] for p in pieces]
    embeddings = llm.embeddings.create(model=EMBED_MODEL, input=chunks)
    store.upsert(
        ids=[f"{doc_id}-{i}" for i in range(len(chunks))],
        documents=chunks,
        embeddings=[e.embedding for e in embeddings.data],
        metadatas=[
            {
                "source": source,
                "doc_id": doc_id,
                "acl": acl,
                "heading": pieces[i]["heading"],
                "content_hash": digest,
                "embed_model": EMBED_MODEL,
            }
            for i in range(len(chunks))
        ],
    )
    return len(chunks)

That is a type-2 rebuild on change: delete old chunks, write new ones, same doc_id. Nightly full re-embeds of a wiki are how teams burn the embedding budget. Hash first.

Query path: retrieve, then generate

def ask(question: str, source: str | None = None, acl: str = "public") -> dict:
    q_emb = llm.embeddings.create(model=EMBED_MODEL, input=[question]).data[0].embedding
    where = {"acl": acl}
    if source:
        where["source"] = source
    hits = store.query(
        query_embeddings=[q_emb],
        n_results=5,
        where=where,
        include=["documents", "metadatas", "distances"],
    )
    docs = hits["documents"][0]
    metas = hits["metadatas"][0]
    context = "\n\n".join(
        f"[{m['source']} / {m.get('heading', '')}]\n{d}" for d, m in zip(docs, metas)
    )
    answer = llm.chat.completions.create(
        model="gpt-5-mini",
        messages=[{
            "role": "user",
            "content": (
                "Answer using only this context. Cite [source / heading] "
                "after each claim. If the context is missing the answer, say so.\n\n"
                f"{context}\n\nQuestion: {question}"
            ),
        }],
    )
    return {
        "answer": answer.choices[0].message.content,
        "sources": metas,
    }

Metadata is your WHERE clause. “Answer from the finance wiki only” beats a better embedding model. ACL filters belong in the retrieve call, not in a prompt that says “please do not use HR docs.” Prompts are not access control.

If you need higher precision after retrieve, add a cheap rerank step on the top 20 and keep the top 5 for the prompt. Do not stuff 40 chunks into the generator — you pay tokens and the model starts ignoring the middle.

Freshness, deletes, and drift

Three jobs, not one:

Incremental ingest — sources with updated_at or a changelog. Re-hash and upsert.
Tombstones — doc_ids present in the index, absent in source. Delete those chunks the same day.
Model rebuild — when embed_model changes, rebuild the whole collection under a new name, flip a pointer, drop the old one. Same pattern as a blue/green table swap.

Track lag like any other pipeline: max(source.updated_at) - max(index.loaded_at). If the wiki can change a policy page and the bot still quotes last month, that is a freshness bug, not an “AI” bug.

flowchart TD
  src[Source of truth] --> inc[Incremental hash + upsert]
  src --> tomb[Tombstone: in index, not in source]
  model[embed_model change] --> swap[New collection, flip pointer]
  inc --> idx[Vector index]
  tomb --> idx
  swap --> idx

Three jobs. Incremental, deletes, and a blue/green rebuild when the embedder changes.

Evals: dbt tests for retrieval

Keep a golden set of 30–80 questions with known doc_ids (and, better, chunk ids). On every ingest or chunker change, measure:

Metric	What it catches
Recall@5	Gold doc missing from retrieve
Recall@1	Ranking is wrong even when the doc is in the index
Citation overlap	Generator ignored the retrieved text
ACL leak rate	A “finance-only” user retrieved an HR chunk
Freshness SLA	Updated gold docs missing after the job

def recall_at_k(gold: list[dict], k: int = 5) -> float:
    hits = 0
    for row in gold:
        result = store.query(
            query_embeddings=[
                llm.embeddings.create(model=EMBED_MODEL, input=[row["q"]]).data[0].embedding
            ],
            n_results=k,
            where={"acl": row["acl"]},
            include=["metadatas"],
        )
        doc_ids = {m["doc_id"] for m in result["metadatas"][0]}
        hits += int(row["doc_id"] in doc_ids)
    return hits / len(gold)

If recall@5 is 0.6, do not spend a week on prompt wording. Fix chunking, metadata, or the source extract. The generator cannot cite a chunk it never saw.

Faithfulness is the second eval: take the answer, take the retrieved text, and score whether each claim appears in context. A cheap model can do that as a batch job. Log both scores next to the DAG run, the same way you log row counts.

Cost that actually shows up

Embedding cost scales with changed characters, not with questions. Generation cost scales with questions × chunks in the prompt.

Controls that work:

Hash-skip unchanged docs
Cap n_results and prompt size
Use a small model for retrieve-side rerank; reserve the larger model for the final answer
Cache frequent questions (FAQ, onboarding) — same pattern as the LLM pipeline cache

A nightly full re-embed of 100k pages is how a prototype becomes a finance conversation. Incremental is the default.

Pitfalls I see in reviews

Chunking mid-table. A policy table split across three chunks retrieves as three half-answers. Keep tables together or turn each row into its own chunk with the table title in metadata.

No ACL on retrieve. If the index holds two tenants, filter in the query. “The model will be careful” is not a control.

Mixing embedding models. Half the rows at 1536 dimensions, half at 768, and a silent quality drop. Store embed_model and refuse the query if it does not match.

Evaluating only the chat UI. A pretty demo with five cherry-picked questions is not a test. Golden doc_ids, run in CI.

Treating the vector store as the source of truth. It is an index. The wiki, the ticket system, or the warehouse table is the source. If they disagree, rebuild from source.

Skipping deletes. Stale SOP chunks outrank the new SOP because they are still in the index. Tombstone job, every run.

A first production slice

Do this in one week, not a platform rewrite:

Pick one source (one Confluence space, or one docs/ folder).
Write extract + hash + heading chunk + embed into a collection.
Hand-label 25 questions with the doc_id you expect.
Measure recall@5. If it is below ~0.8, fix chunks before you add a UI.
Add source and ACL filters to retrieve.
Schedule the ingest. Add the tombstone check.
Only then wire a chat box.

Orchestrate it with the scheduler you already run — the Airflow tutorial is the same DAG shape: extract, transform, load, test.

FAQ

Do I need a vector database on day one? No. A warehouse column plus nearest-neighbor, or a local Chroma index, is enough to measure recall@5. Buy a vector product when you have a freshness SLO and more than one retrieve path.

Should I re-embed the whole wiki every night? Only if the embed model changed. Hash the content. Unchanged docs skip the API call.

Where do ACL filters belong? In the retrieve where clause. A prompt that says “ignore HR docs” is not access control.

What if recall@5 is 0.6? Fix chunking, metadata, or the extract. Do not spend a week rewriting the generator prompt.

Can I mix two embedding models in one collection? No. Different dimensions, different spaces. Store embed_model and rebuild under a new name.

Is the vector store the source of truth? No. It is an index. If it disagrees with the wiki, rebuild from the wiki.

What this means for data engineers

Companies staffing “AI teams” keep discovering the bottleneck is not prompt wording. It is document pipelines, freshness, and access control. That is data engineering. Version the embeddings. Test retrieval. Treat the generator as a flaky API on top of an index you own.