Phase 2: Enterprise Data Engineering · 60 min · Python · Apache Airflow · PostgreSQL
The Problem
Before you can build a RAG system, you need to get data from the client's source systems into a form your pipeline can consume. That means ETL/ELT pipelines: extracting from databases, APIs, and file systems; transforming and cleaning; loading into a staging area. Airflow is the industry standard for orchestrating these pipelines.
ETL vs ELT: choosing the right approach
ETL vs ELT decision framework:
- ETL (Extract, Transform, Load)
- Pattern: Extract from source → Transform in staging → Load to warehouse
- Use when
- Source system can't handle analytical queries (OLTP protection)
- Transformations are complex and need dedicated compute
- Data volume is small enough that staging doesn't cost much
- Client has an existing ETL tool (Informatica, SSIS, Talend)
- Example: Extract from SAP via API → Transform in Python (clean, enrich, deduplicate) → Load to Snowflake
- ELT (Extract, Load, Transform)
- Pattern: Extract from source → Load raw to warehouse → Transform in warehouse
- Use when
- Warehouse has abundant compute (Snowflake, BigQuery, Databricks)
- Source system can handle the extraction load
- You want raw data preserved for re-processing (medallion architecture)
- Transformations are SQL-based and benefit from warehouse MPP
- Example: Extract from PostgreSQL via CDC → Load raw to Snowflake → Transform with dbt (silver/gold layers)
Building airflow DAGs with TaskFlow API
# Airflow DAG using TaskFlow API (Python decorators)
from airflow.decorators import dag, task
from datetime import datetime, timedelta
@dag(
dag_id="rag_data_pipeline",
schedule="0 2 * * *", # Daily at 2 AM
start_date=datetime(2026, 1, 1),
catchup=False,
default_args={
"retries": 3,
"retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True,
"max_retry_delay": timedelta(minutes=30),
"sla": timedelta(hours=3), # Must complete by 5 AM
},
tags=["rag", "data-pipeline", "c4"],
)
def rag_data_pipeline():
"""Daily pipeline: extract from SharePoint, transform, load to vector store."""
@task
def extract_sharepoint_docs(**context):
"""Extract documents from SharePoint via Microsoft Graph API."""
execution_date = context["ds"] # YYYY-MM-DD
# Incremental: only docs modified since last run
docs = graph_client.list_documents(
site_id="acme.sharepoint.com",
modified_since=execution_date,
)
return {"extracted_count": len(docs), "execution_date": execution_date}
@task
def transform_docs(extract_result):
"""Clean, chunk, and prepare documents for embedding."""
from chunker import semantic_chunk
chunks = []
for doc in extract_result["docs"]:
doc_chunks = semantic_chunk(
text=doc["content"],
max_tokens=512,
overlap_tokens=50,
)
for i, chunk in enumerate(doc_chunks):
chunks.append({
"chunk_id": f"{doc['id']}_chunk_{i}",
"content": chunk,
"department": doc.get("department", "unknown"),
"chunk_index": i,
})
# Quality checks
empty_chunks = [c for c in chunks if not c["content"].strip()]
if empty_chunks:
raise ValueError(f"Found {len(empty_chunks)} empty chunks — check extraction")
return {"total_chunks": len(chunks), "chunks": chunks}
@task
def load_to_vector_store(transform_result):
"""Embed chunks and load into pgvector with idempotent upsert."""
import psycopg2
conn = psycopg2.connect("postgresql://user:pass@db:5432/rag_db")
cur = conn.cursor()
# Idempotent: ON CONFLICT means re-running the DAG won't create duplicates
cur.execute("""
INSERT INTO document_chunks (chunk_id, content, embedding, department)
SELECT chunk_id, content, embed(content), department
FROM staging_chunks
ON CONFLICT (chunk_id) DO UPDATE
SET content = EXCLUDED.content,
embedding = EXCLUDED.embedding,
modified_at = NOW();
""")
inserted = cur.rowcount
conn.commit()
conn.close()
return {"inserted_rows": inserted, "status": "success"}
@task
def run_data_quality_checks(load_result):
"""Post-load data quality validation — fails the DAG if checks don't pass."""
import psycopg2
conn = psycopg2.connect("postgresql://user:pass@db:5432/rag_db")
cur = conn.cursor()
checks = []
# Check 1: No null embeddings
cur.execute("SELECT COUNT(*) FROM document_chunks WHERE embedding IS NULL")
null_count = cur.fetchone()[0]
checks.append({"check": "no_null_embeddings", "passed": null_count == 0})
# Check 2: No duplicate chunk_ids
cur.execute("SELECT chunk_id, COUNT(*) FROM document_chunks GROUP BY chunk_id HAVING COUNT(*) > 1")
dups = cur.fetchall()
checks.append({"check": "no_duplicates", "passed": len(dups) == 0})
# Check 3: Row count sanity
cur.execute("SELECT COUNT(*) FROM document_chunks")
total = cur.fetchone()[0]
checks.append({"check": "row_count_sane", "passed": total > 0, "value": total})
conn.close()
failed = [c for c in checks if not c["passed"]]
if failed:
raise ValueError(f"Data quality checks failed: {failed}")
return {"all_passed": True, "checks": checks}
# Define task dependencies: extract → transform → load → quality checks
extract_result = extract_sharepoint_docs()
transform_result = transform_docs(extract_result)
load_result = load_to_vector_store(transform_result)
run_data_quality_checks(load_result)
dag = rag_data_pipeline()
Pipeline reliability patterns
Reliability patterns for production pipelines:
- Retries with backoff
- Problem: Transient failures: API rate limits, network timeouts, DB connection drops
- Solution: retry=3, retry_delay=5min, exponential_backoff=True, max_retry_delay=30min
- Airflow config: default_args={'retries': 3, 'retry_delay': timedelta(minutes=5), 'retry_exponential_backoff': True}
- SLA monitoring
- Problem: Pipeline runs too long and delays downstream consumers
- Solution: Set SLA on DAG. Airflow sends alert when task exceeds SLA.
- Airflow config: default_args={'sla': timedelta(hours=3)}
- Dead letter queue
- Problem: Some records fail transformation but shouldn't block the entire pipeline
- Solution: Write failed records to a dead-letter table for manual review
- Implementation: try/except per record, failed records go to dlq_table
- Idempotency
- Problem: Re-running the pipeline creates duplicates
- Solution: Use ON CONFLICT (upsert) instead of INSERT
- SQL: INSERT INTO ... ON CONFLICT (chunk_id) DO UPDATE SET ...
- Incremental loading
- Problem: Full reload takes hours; only a few documents changed
- Solution: Track watermark (last_modified_date). Only extract documents modified since last run.
- Implementation: Store last_watermark in Airflow Variable. Update after successful load.
- Data quality gates
- Problem: Bad data gets loaded and corrupts the vector store
- Solution: Post-load validation task that checks for nulls, duplicates, and schema compliance
- Implementation: run_data_quality_checks() task after load. Raise exception if checks fail.
Incremental loading patterns
Incremental loading: CDC vs watermark vs full refresh:
- CDC (Change Data Capture)
- How: Database logs every INSERT/UPDATE/DELETE. You consume the log stream.
- Use when: Source DB supports CDC (PostgreSQL logical replication, MySQL binlog)
- Pros
- Near real-time
- Captures deletes
- Low source impact
- Cons
- Complex setup
- Schema changes can break CDC
- Tool: Debezium → Kafka → Airflow sensor
- Watermark column
- How: Track last_modified_at. Query: SELECT * FROM table WHERE last_modified_at > :watermark
- Use when: Source has a reliable updated_at timestamp column
- Pros
- Simple to implement
- Works with any database
- Easy to debug
- Cons
- Misses hard deletes
- Requires indexed timestamp column
- Tool: Airflow Variable storing last watermark, updated after each run
- Full refresh
- How: Truncate and reload entire table every run
- Use when: Table is small (<100K rows) or source doesn't support CDC/watermark
- Pros
- Simplest
- Always consistent
- No state to manage
- Cons
- Slow for large tables
- High source impact
- Tool: TRUNCATE staging; INSERT INTO staging SELECT * FROM source
- Merge (upsert)
- How: Load to staging, then MERGE into target: update existing, insert new
- Use when: You need to update existing records and insert new ones in one operation
- SQL: MERGE INTO target t USING staging s ON t.id = s.id WHEN MATCHED THEN UPDATE ... WHEN NOT MATCHED THEN INSERT ...