Data Engineer

SCD Type 2 Implementation Pattern:

-- SCD Type 2 MERGE in SQL
MERGE INTO dim_customer AS target
USING staging_customer AS source
ON target.customer_id = source.customer_id AND target.is_current = TRUE
WHEN MATCHED AND (
    target.email != source.email OR target.city != source.city
) THEN UPDATE SET
    target.is_current = FALSE,
    target.valid_to = CURRENT_DATE
WHEN NOT MATCHED THEN INSERT (
    customer_id, email, city, valid_from, valid_to, is_current
) VALUES (
    source.customer_id, source.email, source.city, CURRENT_DATE, NULL, TRUE
);
-- Then insert updated records as new rows
INSERT INTO dim_customer (customer_id, email, city, valid_from, valid_to, is_current)
SELECT source.customer_id, source.email, source.city, CURRENT_DATE, NULL, TRUE
FROM staging_customer source JOIN dim_customer target
  ON target.customer_id = source.customer_id AND target.is_current = FALSE
  AND target.valid_to = CURRENT_DATE;

-- ROW_NUMBER: unique sequential number per partition
SELECT
  order_id, customer_id, order_date, amount,
  ROW_NUMBER() OVER (PARTITION BY customer_id ORDER BY order_date DESC) AS order_rank
FROM orders;
-- Get each customer's most recent order: WHERE order_rank = 1

-- RANK vs DENSE_RANK
-- RANK: 1, 2, 2, 4  (skips 3 after tie)
-- DENSE_RANK: 1, 2, 2, 3  (no skip)
SELECT name, salary,
  RANK()       OVER (ORDER BY salary DESC) AS rank,
  DENSE_RANK() OVER (ORDER BY salary DESC) AS dense_rank
FROM employees;

-- LAG and LEAD: access previous/next row value
SELECT
  date, revenue,
  LAG(revenue, 1, 0)  OVER (ORDER BY date) AS prev_day_revenue,
  LEAD(revenue, 1, 0) OVER (ORDER BY date) AS next_day_revenue,
  revenue - LAG(revenue) OVER (ORDER BY date) AS day_over_day_change
FROM daily_metrics;

-- NTILE: divide into n buckets
SELECT customer_id, total_spend,
  NTILE(4) OVER (ORDER BY total_spend DESC) AS spend_quartile  -- 1=top 25%
FROM customer_summary;

-- SUM/AVG with frames: running total
SELECT date, revenue,
  SUM(revenue) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total,
  AVG(revenue) OVER (ORDER BY date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS rolling_7d_avg
FROM daily_revenue;

-- FIRST_VALUE / LAST_VALUE
SELECT product_id, sale_date, price,
  FIRST_VALUE(price) OVER (PARTITION BY product_id ORDER BY sale_date) AS first_price,
  LAST_VALUE(price)  OVER (PARTITION BY product_id ORDER BY sale_date
                           ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS latest_price
FROM price_history;

-- Standard CTE: readable, can be referenced multiple times
WITH monthly_revenue AS (
  SELECT DATE_TRUNC('month', order_date) AS month,
         SUM(amount) AS revenue
  FROM orders GROUP BY 1
),
revenue_with_growth AS (
  SELECT month, revenue,
    LAG(revenue) OVER (ORDER BY month) AS prev_month_revenue,
    (revenue - LAG(revenue) OVER (ORDER BY month)) / LAG(revenue) OVER (ORDER BY month) * 100 AS growth_pct
  FROM monthly_revenue
)
SELECT * FROM revenue_with_growth WHERE month >= '2024-01-01';

-- Recursive CTE: organizational hierarchy
WITH RECURSIVE org_hierarchy AS (
  -- Anchor: start with the CEO (no manager)
  SELECT employee_id, name, manager_id, 0 AS level
  FROM employees WHERE manager_id IS NULL

  UNION ALL

  -- Recursive: join employees to their managers
  SELECT e.employee_id, e.name, e.manager_id, h.level + 1
  FROM employees e
  JOIN org_hierarchy h ON e.manager_id = h.employee_id
)
SELECT * FROM org_hierarchy ORDER BY level, name;

-- PIVOT: rows to columns
SELECT *
FROM (SELECT month, category, revenue FROM sales)
PIVOT (SUM(revenue) FOR category IN ('Electronics', 'Clothing', 'Food'));

-- JSON operations (PostgreSQL)
SELECT
  user_id,
  metadata->>'email' AS email,           -- text extraction
  (metadata->'preferences'->>'theme')    -- nested access
FROM users WHERE metadata ? 'email';

-- Deduplication using ROW_NUMBER
DELETE FROM orders WHERE order_id IN (
  SELECT order_id FROM (
    SELECT order_id,
      ROW_NUMBER() OVER (PARTITION BY external_id ORDER BY created_at DESC) AS rn
    FROM orders
  ) t WHERE rn > 1
);

-- Running user cohort retention (classic DE interview question)
WITH cohorts AS (
  SELECT user_id, DATE_TRUNC('month', first_purchase) AS cohort_month
  FROM (SELECT user_id, MIN(order_date) AS first_purchase FROM orders GROUP BY 1) t
),
activity AS (
  SELECT o.user_id, c.cohort_month,
    DATEDIFF('month', c.cohort_month, o.order_date) AS period_number
  FROM orders o JOIN cohorts c USING(user_id)
)
SELECT cohort_month, period_number,
  COUNT(DISTINCT user_id) AS active_users
FROM activity
GROUP BY 1, 2 ORDER BY 1, 2;

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from delta.tables import DeltaTable

spark = SparkSession.builder.appName("ETL").config("spark.sql.extensions",
    "io.delta.sql.DeltaSparkSessionExtension").getOrCreate()

# Read data
df = spark.read.format("parquet").option("mergeSchema", "true").load("s3a://bucket/raw/orders/")

# Transformations
df_clean = (df
  .filter(F.col("amount") > 0)
  .withColumn("order_date", F.to_date("order_timestamp"))
  .withColumn("year_month", F.date_format("order_date", "yyyy-MM"))
  .withColumnRenamed("cust_id", "customer_id")
  .dropDuplicates(["order_id"])
  .na.fill({"discount": 0, "promo_code": "NONE"})
)

# Aggregation
monthly_revenue = df_clean.groupBy("year_month", "category").agg(
  F.sum("amount").alias("total_revenue"),
  F.count("order_id").alias("order_count"),
  F.countDistinct("customer_id").alias("unique_customers"),
  F.avg("amount").alias("avg_order_value"),
)

# Window functions in Spark
window_spec = Window.partitionBy("customer_id").orderBy("order_date")
df_with_rank = df_clean.withColumn("order_rank", F.row_number().over(window_spec))
df_first_order = df_with_rank.filter(F.col("order_rank") == 1)

# Delta Lake MERGE (upsert)
delta_table = DeltaTable.forPath(spark, "s3a://bucket/silver/orders/")
delta_table.alias("target").merge(
  df_clean.alias("source"),
  "target.order_id = source.order_id"
).whenMatchedUpdate(set={
  "amount": "source.amount",
  "status": "source.status",
  "updated_at": "current_timestamp()"
}).whenNotMatchedInsertAll().execute()

# Write with partitioning
df_clean.write.format("delta").mode("overwrite") \
  .partitionBy("year_month") \
  .option("overwriteSchema", "true") \
  .save("s3a://bucket/silver/orders/")

# Adaptive Query Execution (Spark 3.0+) — enable always
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# Broadcast join for small tables
df_orders.join(F.broadcast(df_small_reference), "product_id")

# Cache frequently used DataFrames
df_customers.cache()   # or .persist(StorageLevel.MEMORY_AND_DISK)
df_customers.count()   # trigger materialization

# Delta Lake OPTIMIZE + ZORDER for query performance
spark.sql("OPTIMIZE silver.orders ZORDER BY (customer_id, order_date)")
spark.sql("VACUUM silver.orders RETAIN 168 HOURS")  # 7-day retention

from kafka import KafkaProducer, KafkaConsumer
import json, time

# Producer — send events
producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8') if k else None,
    acks='all',              # wait for all replicas
    enable_idempotence=True,  # exactly-once
    compression_type='snappy',
)

event = {"user_id": "u_123", "event": "purchase", "amount": 59.99, "ts": time.time()}
future = producer.send("user-events", key=event["user_id"], value=event)
record_metadata = future.get(timeout=10)  # blocks until ack received
print(f"Sent to partition {record_metadata.partition} @ offset {record_metadata.offset}")

producer.flush()
producer.close()

# Consumer — process events
consumer = KafkaConsumer(
    "user-events",
    bootstrap_servers=['kafka:9092'],
    group_id="analytics-pipeline",
    auto_offset_reset="earliest",   # replay from beginning on new consumer group
    enable_auto_commit=False,       # manual offset commit for at-least-once
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    max_poll_records=100,
)

for message in consumer:
    event = message.value
    # Process event
    process_event(event)
    # Commit offset AFTER successful processing
    consumer.commit()

from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
    "retry_exponential_backoff": True,
    "on_failure_callback": alert_slack,  # Slack alert on failure
}

@dag(
    dag_id="orders_etl_pipeline",
    schedule="0 6 * * *",          # 6 AM daily
    start_date=datetime(2024, 1, 1),
    catchup=False,                   # don't backfill missed runs
    default_args=default_args,
    tags=["orders", "production"],
)
def orders_etl():

    @task()
    def extract_from_postgres(**context):
        """Extract yesterday's orders from OLTP database"""
        hook = PostgresHook(postgres_conn_id="prod_postgres")
        execution_date = context["ds"]   # YYYY-MM-DD of current run
        df = hook.get_pandas_df(
            f"SELECT * FROM orders WHERE DATE(created_at) = '{execution_date}'"
        )
        # Push to XCom (auto-handled by @task)
        return df.to_dict("records")

    @task()
    def transform(records):
        """Clean and transform records"""
        import pandas as pd
        df = pd.DataFrame(records)
        df["amount"] = df["amount"].astype(float)
        df = df[df["status"] != "cancelled"]
        df["processed_at"] = pd.Timestamp.now()
        return df.to_dict("records")

    @task()
    def load_to_s3(records, **context):
        """Write Parquet to S3"""
        import pandas as pd, io
        df = pd.DataFrame(records)
        s3_hook = S3Hook(aws_conn_id="aws_default")
        buffer = io.BytesIO()
        df.to_parquet(buffer, index=False)
        s3_hook.load_bytes(buffer.getvalue(),
            key=f"orders/date={context['ds']}/orders.parquet",
            bucket_name="my-data-lake", replace=True)

    # DAG structure: extract → transform → load
    raw = extract_from_postgres()
    clean = transform(raw)
    load_to_s3(clean)

orders_etl()

{{ -- models/silver/stg_orders.sql -- }}
{{
  config(
    materialized = 'incremental',
    unique_key = 'order_id',
    incremental_strategy = 'merge',    -- 'merge' for upserts, 'append' for inserts only
    on_schema_change = 'sync_all_columns'
  )
}}

WITH source AS (
  SELECT * FROM {{ source('raw', 'orders') }}
  {% if is_incremental() %}
  -- Only process records newer than our latest
  WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
  {% endif %}
),
cleaned AS (
  SELECT
    order_id,
    customer_id,
    CAST(amount AS NUMERIC(12,2)) AS amount,
    LOWER(status) AS status,
    DATE(created_at) AS order_date,
    created_at,
    COALESCE(updated_at, created_at) AS updated_at
  FROM source
  WHERE amount > 0
    AND order_id IS NOT NULL
)
SELECT * FROM cleaned

# schema.yml — define tests, sources, documentation
version: 2

sources:
  - name: raw
    database: my_database
    schema: raw_schema
    tables:
      - name: orders
        loaded_at_field: created_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}

models:
  - name: stg_orders
    description: "Cleaned and standardized orders from OLTP"
    columns:
      - name: order_id
        description: "Unique order identifier"
        tests:
          - not_null
          - unique
      - name: amount
        tests:
          - not_null
          - accepted_range:
              min_value: 0
              max_value: 100000
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'completed', 'refunded', 'cancelled']