Why Retail Firms Hire More Data Engineers in 2026

# PySpark — a realistic retail pipeline
# (ingesting POS + e-commerce events into a unified sales table)

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, to_date, sum as _sum, count, lit, coalesce
)

spark = SparkSession.builder.appName("retail_sales_etl").getOrCreate()

# Ingest POS sales (from an on-prem database, dumped to S3 nightly)
pos = spark.read.parquet("s3://my-lake/raw/pos-sales/date=2026-11-01/")

# Ingest e-commerce orders (from Kafka, dumped as parquet)
ecom = spark.read.parquet("s3://my-lake/raw/ecom-orders/date=2026-11-01/")

# Normalize both to a common schema
pos_norm = (pos
    .select(
        col("transaction_id").alias("order_id"),
        col("store_id").alias("channel_id"),
        lit("pos").alias("channel"),
        to_date("transaction_date").alias("order_date"),
        col("sku").alias("product_id"),
        col("quantity"),
        col("amount").alias("revenue"),
    )
)

ecom_norm = (ecom
    .select(
        col("order_id"),
        col("marketplace").alias("channel_id"),
        lit("ecom").alias("channel"),
        to_date("created_at").alias("order_date"),
        col("sku").alias("product_id"),
        col("qty").alias("quantity"),
        col("total").alias("revenue"),
    )
)

# Union and aggregate daily by product + channel
combined = pos_norm.union(ecom_norm)

daily_sales = (combined
    .groupBy("order_date", "product_id", "channel")
    .agg(
        _sum("revenue").alias("revenue"),
        _sum("quantity").alias("units_sold"),
        count("order_id").alias("order_count"),
    )
)

# Write to the warehouse (Redshift/Snowflake/BigQuery)
(daily_sales
    .write
    .mode("overwrite")
    .partitionBy("order_date")
    .parquet("s3://my-lake/marts/daily_sales/")
)

# Skills: multi-source ingestion, schema normalisation, aggregation,
# partitioned writes — the bread and butter of retail data engineering.

-- dbt model: retail customer cohorts
-- (a common analytics engineering task at retail companies)

{{ config(
    materialized='incremental',
    unique_key='cohort_key',
    partition_by={'field': 'cohort_month', 'data_type': 'date'}
) }}

with customers as (
    select
        customer_id,
        min(order_date) as first_order_date,
        date_trunc('month', min(order_date)) as cohort_month
    from {{ ref('stg_orders') }}
    group by customer_id
),

order_activity as (
    select
        o.customer_id,
        c.cohort_month,
        date_trunc('month', o.order_date) as activity_month,
        count(distinct o.order_id) as orders,
        sum(o.revenue) as revenue
    from {{ ref('stg_orders') }} o
    join customers c using (customer_id)
    group by 1, 2, 3
),

final as (
    select
        cohort_month,
        activity_month,
        datediff('month', cohort_month, activity_month) as month_number,
        count(distinct customer_id) as customers_active,
        sum(revenue) as revenue,
        -- Retention: active customers / cohort size
        round(
            100.0 * count(distinct customer_id) /
            first_value(count(distinct customer_id)) over (
                partition by cohort_month
                order by activity_month
            ),
            2
        ) as retention_pct
    from order_activity
    group by 1, 2, 3
)

select * from final

-- This cohort analysis drives retention campaigns,
-- one of the most important use cases in retail.
-- Skills: dbt, window functions, cohort logic, business insight.