How Civil Engineers Can Enter Data Engineering in 2026

# etl_pipeline.py — a real ETL pipeline a junior data engineer builds
# This is the kind of project that impresses interviewers.

import pandas as pd
import requests
from sqlalchemy import create_engine
from datetime import datetime, timedelta

# 1. Extract — pull data from a REST API
def extract(days_back: int = 7) -> pd.DataFrame:
    since = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
    url = f"https://api.example.com/orders?since={since}"
    response = requests.get(url, timeout=30)
    response.raise_for_status()
    return pd.DataFrame(response.json()["orders"])

# 2. Transform — clean, validate, and model the data
def transform(df: pd.DataFrame) -> pd.DataFrame:
    # Drop duplicates
    df = df.drop_duplicates(subset=["order_id"])

    # Parse dates and types
    df["created_at"] = pd.to_datetime(df["created_at"])
    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    # Drop rows with missing critical fields
    df = df.dropna(subset=["order_id", "created_at", "amount"])

    # Add derived columns
    df["order_date"] = df["created_at"].dt.date
    df["month"] = df["created_at"].dt.to_period("M").astype(str)

    # Validate: reject negative amounts
    df = df[df["amount"] >= 0]

    return df[[
        "order_id", "user_id", "order_date", "month",
        "amount", "status", "created_at"
    ]]

# 3. Load — write to a warehouse (PostgreSQL example)
def load(df: pd.DataFrame) -> None:
    engine = create_engine("postgresql://user:pass@localhost:5432/analytics")
    df.to_sql(
        "orders",
        engine,
        if_exists="append",
        index=False,
        method="multi",
    )

# Orchestrate the pipeline
def run():
    raw = extract()
    clean = transform(raw)
    load(clean)
    print(f"Loaded {len(clean)} rows")

if __name__ == "__main__":
    run()

# What this shows an interviewer:
# - You can hit an API, handle errors, validate data, and write to a warehouse
# - You think about idempotency (deduplication by order_id)
# - You split extract / transform / load, not one giant function
# - This is the shape of nearly every real data pipeline

# What civil engineers tend to do well here:
# - You already think about tolerances (validation)
# - You already think about failure modes (retries, deduplication)
# - You already think about load (batch size, memory)
#
# What to practise:
# - Adding logging, retries, and alerting
# - Handling schema drift
# - Writing tests for each stage

# orders_dag.py — an Airflow DAG, the standard orchestration layer
# Every junior data engineer is expected to read and write these.

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

# Import your pipeline functions
from etl_pipeline import extract, transform, load

default_args = {
    "owner": "data-engineering",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-alerts@example.com"],
}

with DAG(
    dag_id="orders_daily_etl",
    start_date=datetime(2026, 1, 1),
    schedule="0 2 * * *",     # every day at 2 AM
    catchup=False,
    default_args=default_args,
    tags=["etl", "orders"],
) as dag:

    extract_task = PythonOperator(
        task_id="extract_orders",
        python_callable=extract,
        op_kwargs={"days_back": 1},
    )

    transform_task = PythonOperator(
        task_id="transform_orders",
        python_callable=transform,
        op_kwargs={"df": extract_task.output},
    )

    load_task = PythonOperator(
        task_id="load_orders",
        python_callable=load,
        op_kwargs={"df": transform_task.output},
    )

    # Simple linear dependency chain
    extract_task >> transform_task >> load_task

# What this shows an interviewer:
# - You understand DAGs (directed acyclic graphs)
# - You know how to configure retries and alerts
# - You can split a pipeline into clear steps
# - You have run something on a schedule, not just ad hoc
#
# Why civil engineers adapt well:
# - A DAG is a project schedule
# - An operator is a task
# - A retry policy is a contingency plan
# You have already thought in these terms on site.

-- dimensional_model.sql — designing the warehouse, not just querying it
-- A critical data engineering skill: modelling data for analytics.

-- 1. Dimension: customers (slowly changing dimension, type 2)
CREATE TABLE dim_customers (
    customer_key     BIGSERIAL PRIMARY KEY,
    customer_id      TEXT NOT NULL,
    email            TEXT,
    city             TEXT,
    signup_date      DATE,
    -- SCD Type 2 columns
    valid_from       TIMESTAMP NOT NULL DEFAULT NOW(),
    valid_to         TIMESTAMP,
    is_current       BOOLEAN NOT NULL DEFAULT TRUE
);

CREATE INDEX idx_customers_id ON dim_customers(customer_id);
CREATE INDEX idx_customers_current ON dim_customers(customer_id) WHERE is_current = TRUE;

-- 2. Fact table: orders
CREATE TABLE fact_orders (
    order_key        BIGSERIAL PRIMARY KEY,
    order_id         TEXT NOT NULL UNIQUE,
    customer_key     BIGINT NOT NULL REFERENCES dim_customers(customer_key),
    order_date       DATE NOT NULL,
    amount           NUMERIC(12, 2) NOT NULL,
    status           TEXT,
    created_at       TIMESTAMP NOT NULL
);

CREATE INDEX idx_orders_date ON fact_orders(order_date);
CREATE INDEX idx_orders_customer ON fact_orders(customer_key);

-- 3. Upsert pattern for idempotent loads
INSERT INTO fact_orders (order_id, customer_key, order_date, amount, status, created_at)
VALUES
  ('ORD-001', 1, '2026-09-10', 1499.00, 'delivered', '2026-09-10 10:22:00'),
  ('ORD-002', 2, '2026-09-10', 2999.00, 'pending',   '2026-09-10 11:45:00')
ON CONFLICT (order_id) DO UPDATE
SET
  amount = EXCLUDED.amount,
  status = EXCLUDED.status;

-- 4. A simple reporting query on the model
SELECT
  d.order_date,
  COUNT(*)                    AS orders,
  SUM(d.amount)               AS revenue,
  ROUND(AVG(d.amount), 2)     AS avg_order_value
FROM fact_orders f
JOIN dim_customers c ON c.customer_key = f.customer_key
WHERE f.order_date >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY 1
ORDER BY 1;

-- Why this matters:
# - Data engineers don't just move data — they model it
# - Star schemas make analytics fast and clear
# - SCD Type 2 is how you track history (customers change cities, emails)
# - Idempotent upserts keep loads safe to re-run
#
# What civil engineers already understand:
# - Normalisation is not unlike standardised unit systems
# - Constraints are like load specifications
# - Referential integrity is like physical consistency