Walmart Data Engineer Interview Questions and Answers 2026

WITH rolling AS (
    SELECT
        store_id,
        week_start,
        total_sales,
        AVG(total_sales) OVER (
            PARTITION BY store_id
            ORDER BY week_start
            ROWS BETWEEN 12 PRECEDING AND 1 PRECEDING
        )  AS rolling_avg_12w,
        STDDEV_SAMP(total_sales) OVER (
            PARTITION BY store_id
            ORDER BY week_start
            ROWS BETWEEN 12 PRECEDING AND 1 PRECEDING
        )  AS rolling_std_12w,
        COUNT(*) OVER (
            PARTITION BY store_id
            ORDER BY week_start
            ROWS BETWEEN 12 PRECEDING AND 1 PRECEDING
        )  AS window_size
    FROM store_weekly_sales
),
zscore AS (
    SELECT
        store_id,
        week_start,
        total_sales,
        ROUND(rolling_avg_12w, 2)                                       AS rolling_avg,
        ROUND(rolling_std_12w, 2)                                       AS rolling_std,
        ROUND((total_sales - rolling_avg_12w)
              / NULLIF(rolling_std_12w, 0), 2)                          AS z_score,
        window_size
    FROM rolling
    WHERE window_size >= 8    -- need at least 8 weeks of history for reliable stats
)
SELECT
    store_id,
    week_start,
    total_sales,
    rolling_avg,
    rolling_std,
    z_score,
    CASE WHEN z_score > 2  THEN 'ABOVE_NORMAL'
         WHEN z_score < -2 THEN 'BELOW_NORMAL'
    END  AS flag
FROM zscore
WHERE ABS(z_score) > 2
ORDER BY week_start DESC, ABS(z_score) DESC;

WITH quarter_cogs AS (
    SELECT
        store_id,
        sku_id,
        SUM(cogs)  AS total_cogs
    FROM sales_transactions
    WHERE txn_date >= DATE_TRUNC('quarter', CURRENT_DATE - INTERVAL '3 months')
      AND txn_date <  DATE_TRUNC('quarter', CURRENT_DATE)
    GROUP BY store_id, sku_id
),
avg_inventory AS (
    SELECT
        store_id,
        sku_id,
        category,
        AVG(inventory_value)  AS avg_inv_value
    FROM inventory_snapshots
    WHERE snapshot_date >= DATE_TRUNC('quarter', CURRENT_DATE - INTERVAL '3 months')
      AND snapshot_date <  DATE_TRUNC('quarter', CURRENT_DATE)
    GROUP BY store_id, sku_id, category
),
turnover AS (
    SELECT
        a.store_id,
        a.sku_id,
        a.category,
        a.avg_inv_value,
        c.total_cogs,
        c.total_cogs / NULLIF(a.avg_inv_value, 0)  AS turnover_ratio
    FROM avg_inventory a
    LEFT JOIN quarter_cogs c USING (store_id, sku_id)
),
with_pctile AS (
    SELECT
        *,
        PERCENTILE_CONT(0.10) WITHIN GROUP (ORDER BY turnover_ratio)
            OVER (PARTITION BY category)  AS p10_turnover
    FROM turnover
)
SELECT
    store_id,
    sku_id,
    category,
    ROUND(avg_inv_value, 2)  AS avg_inv_value,
    ROUND(total_cogs, 2)     AS total_cogs,
    ROUND(turnover_ratio, 3) AS turnover_ratio,
    ROUND(p10_turnover, 3)   AS category_p10,
    'SLOW_MOVING'            AS flag
FROM with_pctile
WHERE turnover_ratio < p10_turnover
   OR turnover_ratio IS NULL   -- zero sales = definitely slow-moving
ORDER BY category, turnover_ratio NULLS FIRST;

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB") \
    .getOrCreate()

# Approach 1: Enable AQE (zero code change, handles skew at runtime)
# AQE splits skewed partitions automatically — best for simple aggregations

df = spark.read.parquet("s3://walmart-datalake/transactions/dt=2026-05-19/")

# Approach 2: Manual salting (when AQE is not enough or Spark < 3.0)
SALT = 20   # number of salt buckets

df_salted = df.withColumn(
    "salted_sku_id",
    F.concat(
        F.col("sku_id"),
        F.lit("_"),
        (F.rand() * SALT).cast("int").cast("string")
    )
)

agg_salted = df_salted.groupBy("store_id", "salted_sku_id", "txn_date") \
    .agg(
        F.sum("revenue").alias("revenue"),
        F.sum("units").alias("units"),
        F.sum("cogs").alias("cogs")
    )

# De-salt: strip the suffix and re-aggregate
agg_final = agg_salted \
    .withColumn("sku_id",
        F.split(F.col("salted_sku_id"), "_").getItem(0)
    ) \
    .groupBy("store_id", "sku_id", "txn_date") \
    .agg(
        F.sum("revenue").alias("total_revenue"),
        F.sum("units").alias("total_units"),
        F.sum("cogs").alias("total_cogs")
    )

agg_final.write.mode("overwrite") \
    .partitionBy("txn_date") \
    .parquet("s3://walmart-datalake/daily_sales_agg/")

from delta.tables import DeltaTable
from pyspark.sql import functions as F
from datetime import datetime

# Load current target and staged changes
target = DeltaTable.forPath(spark, "abfss://warehouse/inventory_history/")
staged = spark.table("stg_inventory_changes")

# Step 1: Close out changed records in target (set valid_to = now, is_current = False)
target.alias("tgt").merge(
    staged.alias("src"),
    "tgt.store_id = src.store_id AND tgt.sku_id = src.sku_id AND tgt.is_current = true"
).whenMatchedUpdate(
    condition = "tgt.inventory_qty != src.inventory_qty OR tgt.unit_cost != src.unit_cost",
    set = {
        "is_current": F.lit(False),
        "valid_to":   F.lit(datetime.utcnow())
    }
).execute()

# Step 2: Insert new records for changed rows (is_current = True)
new_records = staged.join(
    spark.table("inventory_history").filter("is_current = false AND valid_to >= current_timestamp() - interval 1 minute"),
    on=["store_id", "sku_id"],
    how="inner"
).select(
    F.monotonically_increasing_id().alias("surrogate_key"),
    "store_id", "sku_id", "inventory_qty", "unit_cost",
    F.current_timestamp().alias("valid_from"),
    F.lit(None).alias("valid_to"),
    F.lit(True).alias("is_current")
)
new_records.write.format("delta").mode("append").save("abfss://warehouse/inventory_history/")

# Step 3: Update watermark
spark.sql(f"UPDATE pipeline_state SET max_updated_at = '{datetime.utcnow()}' WHERE pipeline = 'inventory_etl'")