DE Interview Questions

WITH ranked AS (
  SELECT *,
         ROW_NUMBER() OVER (PARTITION BY customer_id
                            ORDER BY updated_at DESC) AS rn
  FROM customers_raw
)
SELECT * FROM ranked WHERE rn = 1;

SELECT dept_id, salary
FROM (
  SELECT dept_id, salary,
         DENSE_RANK() OVER (PARTITION BY dept_id ORDER BY salary DESC) AS rnk
  FROM employees
) t
WHERE rnk = 2;

SELECT order_date, revenue,
       SUM(revenue) OVER (ORDER BY order_date) AS running_total,
       AVG(revenue) OVER (ORDER BY order_date
                          ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) AS ma_7d
FROM daily_revenue;

WITH d AS (
  SELECT DISTINCT user_id, login_date FROM logins
), g AS (
  SELECT user_id, login_date,
         login_date - ROW_NUMBER() OVER (PARTITION BY user_id
                                         ORDER BY login_date) * INTERVAL '1 day' AS grp
  FROM d
)
SELECT user_id, MIN(login_date) AS start_day, COUNT(*) AS streak
FROM g GROUP BY user_id, grp
HAVING COUNT(*) >= 3;

-- employees who manage nobody (manager_id contains NULL for the CEO)
SELECT * FROM employees e
WHERE NOT EXISTS (SELECT 1 FROM employees r WHERE r.manager_id = e.employee_id);

SELECT e.first_name, e.salary, m.first_name AS manager, m.salary AS manager_salary
FROM employees e
JOIN employees m ON m.employee_id = e.manager_id
WHERE e.salary > m.salary;

SELECT e.*
FROM employees e
JOIN jobs j ON j.job_id = e.job_id
WHERE e.salary NOT BETWEEN j.min_salary AND j.max_salary;

WITH RECURSIVE org(employee_id, lvl, path) AS (
  SELECT employee_id, 1, first_name FROM employees WHERE manager_id IS NULL
  UNION ALL
  SELECT e.employee_id, o.lvl + 1, o.path || ' > ' || e.first_name
  FROM employees e JOIN org o ON e.manager_id = o.employee_id
)
SELECT * FROM org ORDER BY path;

SELECT department_id, AVG(salary) AS median_salary
FROM (
  SELECT department_id, salary,
         ROW_NUMBER() OVER (PARTITION BY department_id ORDER BY salary) AS rn,
         COUNT(*)     OVER (PARTITION BY department_id)                 AS cnt
  FROM employees
) t
WHERE rn IN ((cnt + 1) / 2, (cnt + 2) / 2)
GROUP BY department_id;

SELECT c.customer_id, o.order_id
FROM customers c
LEFT JOIN orders o ON o.customer_id = c.customer_id
WHERE o.status = 'SHIPPED';

SELECT CASE region WHEN NULL THEN 'missing' ELSE region END FROM sales;

SELECT ts, device_id,
       LAST_VALUE(reading IGNORE NULLS) OVER (
         PARTITION BY device_id ORDER BY ts
         ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS reading_filled
FROM sensor_readings;

SELECT COUNT(*)                                  AS total_rows,
       COUNT(*) - COUNT(email)                   AS null_emails,
       SUM(CASE WHEN phone IS NULL THEN 1 ELSE 0 END) AS null_phones
FROM customers;

 cloud storage (S3 / ADLS / GCS)
   └─ Unity Catalog: storage credential → external location → catalog → schemas → tables / volumes
 sources ──► INGEST (Auto Loader, Kafka streaming, Lakeflow Connect)
   ──► BRONZE (raw, append-only Delta) ──► SILVER (clean, deduplicated, merged) ──► GOLD (marts, aggregates)
   ──► SERVE: SQL warehouses, dashboards, BI tools, ML, Delta Sharing
 cross-cutting: Lakeflow Jobs / Declarative Pipelines · Asset Bundles + CI/CD · grants, masks, lineage · monitoring

CREATE EXTERNAL LOCATION IF NOT EXISTS raw_landing
  URL 's3://acme-lake-prod/landing'
  WITH (STORAGE CREDENTIAL acme_lake_cred);

CREATE CATALOG IF NOT EXISTS fraud_prod
  MANAGED LOCATION 's3://acme-lake-prod/managed/fraud_prod';

CREATE SCHEMA IF NOT EXISTS fraud_prod.bronze;
CREATE SCHEMA IF NOT EXISTS fraud_prod.silver;
CREATE SCHEMA IF NOT EXISTS fraud_prod.gold;
CREATE VOLUME IF NOT EXISTS fraud_prod.bronze.checkpoints;     -- streaming checkpoints and files

from pyspark.sql import functions as F

(spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/Volumes/fraud_prod/bronze/checkpoints/txn_schema")
    .load("s3://acme-lake-prod/landing/transactions/")
    .select("*",
            F.col("_metadata.file_path").alias("source_file"),
            F.current_timestamp().alias("ingested_at"))
    .writeStream
    .option("checkpointLocation", "/Volumes/fraud_prod/bronze/checkpoints/txn")
    .trigger(availableNow=True)                    # incremental batch: process new files, then stop
    .toTable("fraud_prod.bronze.transactions"))

from delta.tables import DeltaTable
from pyspark.sql import Window

def upsert_silver(batch_df, batch_id):
    latest = (batch_df
        .filter("txn_id IS NOT NULL AND amount > 0")
        .withColumn("rn", F.row_number().over(Window.partitionBy("txn_id").orderBy(F.col("ingested_at").desc())))
        .filter("rn = 1").drop("rn"))
    (DeltaTable.forName(spark, "fraud_prod.silver.transactions").alias("t")
        .merge(latest.alias("s"), "t.txn_id = s.txn_id")
        .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute())

(spark.readStream.table("fraud_prod.bronze.transactions")
    .writeStream.foreachBatch(upsert_silver)
    .option("checkpointLocation", "/Volumes/fraud_prod/bronze/checkpoints/silver_txn")
    .trigger(availableNow=True)
    .start())

CREATE OR REPLACE TABLE fraud_prod.gold.daily_channel_summary
CLUSTER BY (txn_date) AS
SELECT txn_date, channel, COUNT(*) AS txns, SUM(amount) AS total_amount,
       COUNT_IF(is_flagged) AS flagged_txns
FROM fraud_prod.silver.transactions
GROUP BY txn_date, channel;

GRANT USE CATALOG ON CATALOG fraud_prod TO `fraud-analysts`;
GRANT USE SCHEMA, SELECT ON SCHEMA fraud_prod.gold TO `fraud-analysts`;

-- column mask: only the pii-readers group sees full card numbers
CREATE FUNCTION fraud_prod.silver.mask_pan(pan STRING) RETURNS STRING
RETURN CASE WHEN is_account_group_member('pii-readers') THEN pan
            ELSE concat('XXXX-XXXX-XXXX-', right(pan, 4)) END;
ALTER TABLE fraud_prod.silver.cards ALTER COLUMN pan SET MASK fraud_prod.silver.mask_pan;

-- row filter: analysts see only their region
CREATE FUNCTION fraud_prod.silver.region_filter(region STRING) RETURNS BOOLEAN
RETURN is_account_group_member('fraud-admins') OR region = 'IN-WEST';
ALTER TABLE fraud_prod.silver.transactions SET ROW FILTER fraud_prod.silver.region_filter ON (region);

s3://lake/silver/orders/_delta_log/
  00000000000000000000.json             ← commit 0 (measured: protocol, metaData, add, commitInfo)
  …
  00000000000000000010.checkpoint.parquet   ← full table state at version 10
  00000000000000000011.json
  _last_checkpoint                       ← points readers to the latest checkpoint

Phase 1: find the touched files
  inner-join SOURCE with TARGET on the ON condition (pruning target files with
  partition values and min/max stats) → the list of target files containing at least one match

Phase 2: rewrite those files
  join SOURCE with only the touched files → for each row apply the clauses:
    matched   → UPDATE / DELETE
    unmatched → INSERT (whenNotMatched)
    untouched rows in touched files → copied unchanged into the new files
  commit: "remove" the touched files + "add" the new files, in ONE atomic commit

DESCRIBE HISTORY silver.orders;                                    -- versions, operations, users
SELECT * FROM silver.orders VERSION AS OF 42;
SELECT * FROM silver.orders TIMESTAMP AS OF '2026-10-30 06:00:00';
RESTORE TABLE silver.orders TO VERSION AS OF 42;                   -- roll back a bad load

spark.read.format("delta").option("versionAsOf", 42).load(path)

VACUUM silver.orders;                      -- default retention (7 days)
VACUUM silver.orders RETAIN 168 HOURS;
VACUUM silver.orders DRY RUN;              -- list what would be deleted

OPTIMIZE silver.transactions;                                    -- compaction only
OPTIMIZE silver.transactions WHERE txn_date >= '2026-10-01';     -- just recent partitions
OPTIMIZE silver.transactions ZORDER BY (account_id, merchant_id);

Debezium / AWS DMS / Lakeflow Connect ──► bronze_orders_cdc (append-only: op, before/after, commit ts)
   ──► dedupe: keep the LATEST change per key in each batch (order by commit sequence)
   ──► MERGE into silver.orders: op = 'D' → delete, otherwise upsert

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

def apply_cdc(batch_df, batch_id):
    latest = (batch_df
        .withColumn("rn", F.row_number().over(
            Window.partitionBy("order_id").orderBy(F.col("commit_lsn").desc())))
        .filter("rn = 1").drop("rn"))                       # one change per key, or MERGE fails
    (DeltaTable.forName(spark, "silver.orders").alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedDelete(condition="s.op = 'D'")
        .whenMatchedUpdateAll(condition="s.commit_lsn > t.commit_lsn")   # ignore stale replays
        .whenNotMatchedInsertAll(condition="s.op <> 'D'")
        .execute())

(spark.readStream.table("bronze.orders_cdc")
    .writeStream.foreachBatch(apply_cdc)
    .option("checkpointLocation", "/chk/silver_orders_cdc")
    .trigger(availableNow=True)
    .start())

ALTER TABLE silver.orders SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

changes = (spark.readStream.format("delta")
    .option("readChangeFeed", "true")
    .table("silver.orders"))
# columns: the row + _change_type (insert / update_preimage / update_postimage / delete),
#          _commit_version, _commit_timestamp

def flatten(d, parent="", sep="."):
    out = {}
    for k, v in d.items():
        key = f"{parent}{sep}{k}" if parent else k
        if isinstance(v, dict):
            out.update(flatten(v, key, sep))
        else:
            out[key] = v
    return out