Stop Building Toy Pipelines: The 2026 Data Engineering Portfolio Guide (with Code)

# data/raw_omnichannel_events.csv
event_id,channel,raw_payload,created_at
evt_001,support_chat,"{""agent"": ""Hello"", ""user"": ""App crashed on iOS 17 after update""}",2026-03-20T10:00:00Z
evt_002,product_review,"⭐⭐⭐⭐⭐ The new dashboard is incredibly fast!",2026-03-20T10:05:00Z
evt_003,email_ticket,"<p>URGENT: Billing failed for invoice #9921. Refund?</p>",2026-03-20T10:15:00Z

# models/intermediate/int_events_tokenized.py
import pyspark.sql.functions as F

def model(dbt, session):
    # 1. PRODUCTION CONFIGURATION (FOR SCALE)
    dbt.config(
        materialized="incremental",
        unique_key="event_id",
        partition_by={"field": "event_date", "data_type": "date"}
    )

    df = dbt.ref("raw_omnichannel_events")

    # 2. INCREMENTAL PROCESSING LOGIC
    if dbt.is_incremental:
        max_date_query = f"SELECT MAX(created_at) FROM {dbt.this}"
        max_date = session.sql(max_date_query).collect()[0][0]
        df = df.filter(F.col("created_at") > F.lit(max_date))

    # 3. COMPUTE OPTIMIZATION (Prevent OOM on 5M rows)
    df = df.repartition(16)

    # 4. OMNICHANNEL PARSING (THE CHAOS HANDLER)
    extracted_df = df.withColumn(
        "extracted_text",
        F.when(
            F.col("channel") == "support_chat",
            F.get_json_object(F.col("raw_payload"), "$.user")
        )
        .when(
            F.col("channel") == "email_ticket",
            F.regexp_replace(F.col("raw_payload"), "<[^>]*>", "") # Strip HTML
        )
        .otherwise(F.col("raw_payload"))
    )

    # Clean and normalize
    cleaned_df = extracted_df.withColumn(
        "clean_text",
        F.lower(F.regexp_replace("extracted_text", "[^\w\s]", ""))
    ).filter(F.col("clean_text") != "na")

    # 5. TOKENIZATION & TAGGING (Save LLM Costs)
    tokenized_df = cleaned_df.withColumn(
        "tokens", F.split(F.col("clean_text"), "\s+")
    ).withColumn("token_count", F.size(F.col("tokens")))

    final_df = tokenized_df.withColumn(
        "issue_type",
        F.when(F.col("clean_text").rlike("crash|error|bug|failed"), "technical_issue")
         .when(F.col("clean_text").rlike("refund|billing|invoice"), "billing_issue")
         .otherwise("general")
    ).withColumn("event_date", F.to_date(F.col("created_at")))

    return final_df.select(
        "event_id", "channel", "clean_text", "tokens", "token_count", "issue_type", "created_at", "event_date"
    )

-- models/semantic/mrt_customer_insights.sql
{{ config(materialized='table') }}

SELECT
    event_id AS customer_event_id,
    channel,
    issue_type,
    token_count,
    clean_text AS ai_ready_context,
    created_at AS timestamp
FROM {{ ref('int_events_tokenized') }}
WHERE token_count > 2 -- Filter out empty noise

# dags/omnichannel_ai_pipeline.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago
from datetime import timedelta
import logging

# 1. ALERTING: The Failure Callback
def slack_alert_on_failure(context):
    """Triggers a Slack webhook if the pipeline fails after all retries."""
    task_instance = context.get('task_instance')
    dag_id = task_instance.dag_id
    task_id = task_instance.task_id
    log_url = task_instance.log_url

    error_message = f"🚨 *DAG Failed!* 🚨\n*DAG:* {dag_id}\n*Task:* {task_id}\n*Logs:* <{log_url}|View Logs>"
    logging.error(f"Executing Slack Alert: {error_message}")
    # In production: requests.post(SLACK_WEBHOOK_URL, json={'text': error_message})

# 2. PRODUCTION DEFAULTS
default_args = {
    'owner': 'data_engineering_team',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 3, # Never fail on the first transient network blip
    'retry_delay': timedelta(minutes=5), # Wait 5 mins for the Spark cluster to recover
    'on_failure_callback': slack_alert_on_failure, # Alert ONLY after 3 retries fail
}

with DAG(
    'omnichannel_ai_preparation_pipeline',
    default_args=default_args,
    description='Ingests, cleans, and prepares 5M+ daily customer records for LLM embeddings.',
    schedule_interval='0 3 * * *', # Run daily at 3:00 AM
    start_date=days_ago(1),
    catchup=False,
    tags=['production', 'ai_platform', 'pyspark'],
) as dag:

    # Task 1: Run the PySpark heavy lifting (Intermediate Layer)
    run_spark_transformations = BashOperator(
        task_id='run_pyspark_cleaning_layer',
        bash_command='dbt run --select models/intermediate/',
        # Specific override: PySpark jobs are heavy, give it a longer timeout
        execution_timeout=timedelta(hours=2)
    )

    # Task 2: Build the Semantic models (SQL)
    run_semantic_layer = BashOperator(
        task_id='run_sql_semantic_layer',
        bash_command='dbt run --select models/semantic/',
    )

    # Task 3: Data Quality Tests
    run_data_tests = BashOperator(
        task_id='run_dbt_data_tests',
        bash_command='dbt test',
    )

    # Define Dependencies
    run_spark_transformations >> run_semantic_layer >> run_data_tests