Data engineer coding interview questions

As asked

You are consuming a Kafka topic of click events. Due to producer retries, some events are duplicated. Each event has a unique event_id (UUID) and an event_timestamp. Write a function that processes a batch of events and returns only the first occurrence of each event, keeping the one with the earliest timestamp if event_ids collide. Assume the batch fits in memory.

Sample answer outline

Group by event_id and keep the entry with the minimum event_timestamp. A dictionary keyed by event_id with the earliest event wins. O(n) time and O(n) space where n is batch size. Strong candidates handle the edge case where two events with the same event_id have identical timestamps (tie-break deterministically), and discuss how this changes when the batch does not fit in memory (use a distributed dedup key store like Redis or a MERGE in the sink).

Reference implementation (python)

from typing import List, Dict
from datetime import datetime

def deduplicate_events(events: List[Dict]) -> List[Dict]:
    """
    events: list of dicts with keys: event_id (str), event_timestamp (datetime), payload (dict)
    Return deduplicated list keeping earliest timestamp per event_id.
    """
    seen: Dict[str, Dict] = {}
    for event in events:
        eid = event["event_id"]
        if eid not in seen or event["event_timestamp"] < seen[eid]["event_timestamp"]:
            seen[eid] = event
    return list(seen.values())

Expect these follow-ups

• How would your approach change if the stream is unbounded and you need to deduplicate across many batches without storing all seen IDs forever?
• What is a Bloom filter and how could it help here?

As asked

Given a table of daily revenue by user, write a SQL query to compute the 7-day rolling average revenue per user for each day, including days where fewer than 7 days of history exist. The table has columns: user_id, event_date (DATE), daily_revenue (NUMERIC).

Sample answer outline

Use a window function with ROWS BETWEEN 6 PRECEDING AND CURRENT ROW to include the current day and 6 prior days. AVG() over that frame handles the fewer-than-7-days case automatically by averaging available rows. Strong candidates specify ORDER BY event_date inside the OVER clause, mention that RANGE BETWEEN INTERVAL '6 days' PRECEDING AND CURRENT ROW handles gaps in dates differently than ROWS BETWEEN, and discuss how to handle NULL revenue days (either fill with 0 or exclude from average).

Reference implementation (sql)

-- Table: user_daily_revenue(user_id, event_date, daily_revenue)
-- Write a query returning:
-- user_id | event_date | daily_revenue | rolling_7d_avg

SELECT
  user_id,
  event_date,
  daily_revenue,
  AVG(daily_revenue) OVER (
    PARTITION BY user_id
    ORDER BY event_date
    ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
  ) AS rolling_7d_avg
FROM user_daily_revenue
ORDER BY user_id, event_date;

Expect these follow-ups

• How does the RANGE BETWEEN syntax differ from ROWS BETWEEN when there are missing dates?
• How would you write this query differently in BigQuery vs Snowflake given their SQL dialect differences?

As asked

You receive Kafka messages where each event is a nested JSON with a top-level 'user' object containing 'id' and 'country', and an 'events' array where each item has 'type' and 'ts'. Write PySpark code to explode and flatten this into a DataFrame with one row per event: user_id, country, event_type, event_ts.

Sample answer outline

Use from_json with a defined schema to parse the JSON string column, access nested fields with dot notation, and use explode() to expand the array into rows. Strong candidates define an explicit StructType schema rather than using schema_of_json inference (which is expensive on large datasets), handle nulls in the events array with explode vs explode_outer, and select the individual struct fields after explode using col notation.

Reference implementation (python)

from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql.functions import from_json, col, explode

schema = StructType([
    StructField("user", StructType([
        StructField("id", StringType()),
        StructField("country", StringType())
    ])),
    StructField("events", ArrayType(StructType([
        StructField("type", StringType()),
        StructField("ts", LongType())
    ])))
])

# raw_df has a column 'value' containing the JSON string
# TODO: parse, explode events, and select the four columns

Expect these follow-ups

• What is the difference between explode and explode_outer in Spark?
• How would you handle a schema where the structure of the 'events' array can vary between messages?

As asked

You have a source table orders in Postgres and a target table in Snowflake. The source grows by insert-only appends. Write the Python and SQL logic to implement a high-watermark incremental load that reads only new rows each run, stores the watermark durably, and is safe to rerun without creating duplicates.

Sample answer outline

The pattern: read the last watermark from a metadata table (or a file), query the source for rows where updated_at > watermark, load them to a staging table, MERGE into target on primary key, then update the watermark to MAX(updated_at) of the loaded batch. Strong candidates note that the watermark update must happen after the successful load (or be transactional), that using >= instead of > with the stored watermark risks missing the last row processed, and that updated_at must be indexed on the source for performance. They also note that insert-only sources can use APPEND only without a MERGE.

Reference implementation (python)

import psycopg2
import snowflake.connector

def get_last_watermark(sf_conn) -> str:
    cur = sf_conn.cursor()
    cur.execute("SELECT MAX(watermark_ts) FROM pipeline_metadata WHERE table_name = 'orders'")
    row = cur.fetchone()
    return row[0] or '1970-01-01 00:00:00'

def extract_new_orders(pg_conn, since_ts: str):
    cur = pg_conn.cursor()
    cur.execute(
        "SELECT id, customer_id, total, updated_at FROM orders WHERE updated_at > %s",
        (since_ts,)
    )
    return cur.fetchall()

# TODO: load rows to Snowflake staging, MERGE into target, update watermark

Expect these follow-ups

• What breaks in this pattern if the source table allows updates to the updated_at column retroactively?
• How would you handle timezone issues when comparing timestamps across Postgres and Snowflake?

As asked

Given a table sales(rep_id, region, amount, sale_date), write a single SQL query that returns the top 3 sales by amount for each region. The result should include all columns. Do not use a correlated subquery; use window functions.

Sample answer outline

Use RANK() or ROW_NUMBER() over (PARTITION BY region ORDER BY amount DESC) in a CTE or subquery, then filter on the rank column in the outer query. ROW_NUMBER() assigns unique ranks even for ties; RANK() assigns the same rank to ties and may skip numbers. For exactly top 3 with ties included, DENSE_RANK() is sometimes appropriate. Strong candidates explain the trade-offs between the three and discuss how the WHERE rank <= 3 clause cannot appear in the same SELECT that defines the window function, requiring a CTE or subquery.

Reference implementation (sql)

-- Table: sales(rep_id INT, region VARCHAR, amount DECIMAL, sale_date DATE)
-- Return top 3 sales by amount per region

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC) AS rn
  FROM sales
)
SELECT rep_id, region, amount, sale_date
FROM ranked
WHERE rn <= 3;

Expect these follow-ups

• What is the difference between RANK, DENSE_RANK, and ROW_NUMBER for handling ties?
• How would you write this query efficiently in BigQuery on a 10-billion-row table?

As asked

You have a staging table stg_customers with fresh data from an API and a target table dim_customers. Write a Snowflake MERGE statement that inserts new customers, updates existing ones only when any column has changed (to avoid unnecessary writes), and sets an updated_at column to the current timestamp on update.

Sample answer outline

The MERGE matches on customer_id. The WHEN MATCHED AND <any column differs> THEN UPDATE clause uses a compound condition comparing each column (or uses a hash of all columns for brevity: MD5 of a JSON object construct) to avoid updates on unchanged rows. The WHEN NOT MATCHED THEN INSERT branch handles new rows. Strong candidates note that checking every column individually is verbose but transparent, while hashing is concise but makes debugging harder. They also mention that unnecessary UPDATEs waste Snowflake credits and inflate historical row counts if using change tracking.

Reference implementation (sql)

MERGE INTO dim_customers AS tgt
USING stg_customers AS src
  ON tgt.customer_id = src.customer_id
WHEN MATCHED AND (
    tgt.email    <> src.email OR
    tgt.country  <> src.country OR
    tgt.plan     <> src.plan
) THEN UPDATE SET
    tgt.email      = src.email,
    tgt.country    = src.country,
    tgt.plan       = src.plan,
    tgt.updated_at = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (customer_id, email, country, plan, updated_at)
  VALUES (src.customer_id, src.email, src.country, src.plan, CURRENT_TIMESTAMP());

Expect these follow-ups

• How would you extend this MERGE to also soft-delete customers that are present in the target but absent from the staging table?
• What is the performance impact of MERGE on a Snowflake table with 500 million rows compared to DELETE+INSERT?

As asked

You have a Spark DataFrame with columns start and end representing time intervals. Write a PySpark transformation that merges all overlapping intervals and returns the minimal set of non-overlapping intervals. For example, [(1,3),(2,6),(8,10),(15,18)] becomes [(1,6),(8,10),(15,18)].

Sample answer outline

The candidate should describe the algorithm: sort by start, then use a window function or fold to track the current maximum end and detect whether each interval merges with the previous one. A clean Spark approach uses Window.orderBy('start') and lag to detect gap boundaries, then groups by boundary and takes min(start)/max(end) per group. Alternatively, the candidate can collectAsList on smaller datasets and apply the linear scan algorithm. They should discuss the tradeoffs between a full Spark SQL approach versus collecting to driver.

Reference implementation (python)

from pyspark.sql import functions as F
from pyspark.sql.window import Window

df = spark.createDataFrame([(1,3),(2,6),(8,10),(15,18)], ["start","end"])

w = Window.orderBy("start")
df2 = df.withColumn(
    "prev_max_end",
    F.max("end").over(w.rowsBetween(Window.unboundedPreceding, -1))
).withColumn(
    "new_group",
    F.when(F.col("start") > F.col("prev_max_end"), 1).otherwise(0)
).withColumn("group_id", F.sum("new_group").over(w))

result = df2.groupBy("group_id").agg(
    F.min("start").alias("start"),
    F.max("end").alias("end")
).orderBy("start")

Expect these follow-ups

• How does your approach change if intervals can come from multiple sources and you need to stream merge them?
• What is the time complexity of the distributed version compared to the single-machine version?

As asked

Write a PySpark job that reads a 1TB text file from cloud storage and returns the top 100 most frequent words, excluding common English stop words. Walk me through your implementation and the performance choices you make.

Sample answer outline

The candidate should split lines into words, lowercase and strip punctuation, filter stop words using a broadcast variable, count occurrences, and use top() or orderBy with limit. Key performance choices: broadcasting the stop word set avoids a shuffle join, using flatMap then countByValue is correct but aggregating with reduceByKey is more scalable than groupByKey. They should use Spark's built-in functions (explode, lower, regexp_replace) for efficiency over Python UDFs.

Reference implementation (python)

from pyspark.sql import functions as F

stop_words = spark.sparkContext.broadcast(
    set(["the", "a", "an", "is", "in", "it", "of", "and", "to"])
)

df = spark.read.text("s3://bucket/large-file.txt")

result = (
    df.select(F.explode(F.split(F.lower(F.regexp_replace("value", "[^a-zA-Z ]", "")), " ")).alias("word"))
    .filter(F.col("word") != "")
    .filter(~F.col("word").isin(list(stop_words.value)))
    .groupBy("word")
    .count()
    .orderBy(F.desc("count"))
    .limit(100)
)

Expect these follow-ups

• Why is groupByKey a bad choice here, and what does reduceByKey do differently?
• If the output had to update incrementally as new files arrive, how would you change the approach?

As asked

You are reviewing Scala code that processes millions of records with chained collection operations. What performance issues do you look for?

Sample answer outline

The candidate should look for accidental intermediate collections, strict versus lazy evaluation, poor choice of List for indexed access, and repeated groupBy or sortBy on large datasets. Iterator or View can avoid intermediate allocation, but they also move work later and can surprise readers if reused incorrectly. Vector is often a better general-purpose immutable sequence than List when indexed access or appends matter. For distributed data, the answer should separate local collection costs from Spark or Akka stream costs. Strong answers avoid premature optimisation but know where idiomatic code becomes expensive.

Reference implementation (scala)

val totalsByUser: Map[String, BigDecimal] =
  events.iterator
    .filter(_.status == "paid")
    .map(event => event.userId -> event.amount)
    .foldLeft(Map.empty[String, BigDecimal]) { case (acc, (userId, amount)) =>
      acc.updated(userId, acc.getOrElse(userId, BigDecimal(0)) + amount)
    }

Expect these follow-ups

• When is List the wrong default?
• How do View and Iterator differ?
• How would you prove a collection change actually helped?

As asked

Write a PySpark word count in both the RDD API and the DataFrame API. Then explain when in 2024 you would still choose RDD over DataFrame, and what performance difference Catalyst optimization makes.

Sample answer outline

RDD API: parallelize, flatMap to split words, map to (word,1), reduceByKey. DataFrame API: read text file, split the column using split and explode, groupBy word, count(). Catalyst optimizes DataFrame queries with rule-based and cost-based rewrites (predicate pushdown, constant folding, join reordering) that RDDs do not get. Tungsten adds code generation and off-heap memory. RDDs are still appropriate in rare cases: working with non-JVM serializable objects that cannot be expressed as DataFrame rows, applying arbitrary Python functions that cross the JVM boundary and cannot be expressed in SQL, or interfacing with low-level Spark internals. Note that the RDD-based MLlib package (spark.mllib) is deprecated as of Spark 3.0 and has been removed in many distributions; the DataFrame-based spark.ml package is the current standard and does not require RDDs. Strong candidates note that the performance gap is significant: DataFrames are typically 10 to 100x faster for typical analytics workloads because of Catalyst and Tungsten.

Reference implementation (python)

# RDD approach
rdd = sc.textFile("words.txt")
word_counts_rdd = (
    rdd.flatMap(lambda line: line.split())
       .map(lambda word: (word, 1))
       .reduceByKey(lambda a, b: a + b)
)

# DataFrame approach
from pyspark.sql.functions import split, explode, col
df = spark.read.text("words.txt")
word_counts_df = (
    df.select(explode(split(col("value"), r"\s+")).alias("word"))
      .groupBy("word")
      .count()
)

Expect these follow-ups

• What does the Catalyst optimizer's logical plan vs physical plan distinction mean?
• When does using a Python UDF in PySpark break Catalyst optimization?

As asked

You have a Snowflake table partitioned by event_date. Rows sometimes arrive 2 to 7 days late. Write a SQL query that identifies which partitions have received new rows in the last 24 hours that belong to a date older than yesterday, helping you decide which downstream dbt models need to be rerun.

Sample answer outline

Snowflake does not automatically track a per-row load timestamp on regular tables. You must add an explicit column, typically named loaded_at or load_time, populated via a DEFAULT CURRENT_TIMESTAMP() on INSERT or via a COPY INTO transformation. With that column in place, the query filters on loaded_at >= CURRENT_TIMESTAMP - INTERVAL '24 hours' AND event_date < CURRENT_DATE - 1 to find late-arriving rows. Alternatively, for tables loaded via COPY INTO, INFORMATION_SCHEMA.LOAD_HISTORY records file-level load timestamps, though it does not give row-level granularity. Group by event_date to see which partitions were affected. Strong candidates mention using Snowflake streams with change tracking enabled as an alternative approach, and discuss how to feed the result into an Airflow sensor that triggers only the affected downstream models rather than a full rerun.

Reference implementation (sql)

-- Identify event_dates that received new rows in last 24h
-- but are older than yesterday (i.e., late arrivals)
SELECT
  event_date,
  COUNT(*) AS late_row_count,
  MAX(load_time) AS latest_load_time
FROM events
WHERE
  load_time >= CURRENT_TIMESTAMP - INTERVAL '24 hours'
  AND event_date < CURRENT_DATE - 1
GROUP BY event_date
ORDER BY event_date DESC;

Expect these follow-ups

• How would you add a loaded_at column to an existing high-volume table without downtime, and what COPY INTO option ensures it is populated automatically?
• How do you prevent late data from causing incorrect results in an already-computed daily aggregate?

As asked

Using the confluent-kafka Python library, write a consumer that reads from a topic, processes each message, and commits offsets only after successful processing. If processing fails, the consumer should log the error, skip the bad message after 3 retries, and continue. Do not use auto-commit.

Sample answer outline

Disable enable.auto.commit, call consumer.poll() in a loop, process the message inside a try/except, track retry counts in a dictionary keyed by partition and offset, commit with consumer.commit(message=msg, asynchronous=False) only on success. After max retries, log to a dead-letter topic or file and commit to advance past the poison-pill message. Strong candidates discuss the risk of committing before processing (at-most-once) versus after (at-least-once), and mention using a dead-letter queue Kafka topic rather than silently skipping.

Reference implementation (python)

from confluent_kafka import Consumer, KafkaException
import logging

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'enable.auto.commit': False,
    'auto.offset.reset': 'earliest',
}
consumer = Consumer(conf)
consumer.subscribe(['my-topic'])

MAX_RETRIES = 3
retry_counts: dict = {}

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            raise KafkaException(msg.error())
        # TODO: process msg.value(), track retries, commit on success
finally:
    consumer.close()

Expect these follow-ups

• What is a dead-letter queue in a Kafka context and how do you set up retry processing for DLQ messages?
• How does your retry logic interact with consumer rebalancing? Could a message be retried more than 3 times across restarts?

As asked

A data scientist handed you a PySpark job that calls a Python UDF on every row to parse a custom date format and categorize revenue buckets. The job runs 5x slower than expected. Walk me through diagnosing and fixing the performance problem.

Sample answer outline

Python UDFs break JVM optimization: each row crosses the Python-JVM boundary via Apache Arrow or byte serialization, disabling Catalyst optimization and Tungsten code generation. Fix 1: replace with built-in Spark SQL functions if possible (to_date, when/otherwise for bucketing). Fix 2: if custom logic is unavoidable, use Pandas UDF (also called vectorized UDF) which processes columnar Arrow batches instead of row-by-row, reducing serialization overhead by 10 to 100x. Fix 3: for complex logic, write a Scala UDF and register it in the SparkSession. Strong candidates demonstrate the Pandas UDF decorator syntax and mention that Pandas UDFs require Arrow enabled via spark.sql.execution.arrow.pyspark.enabled.

Reference implementation (python)

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType
import pandas as pd

# Slow: row-by-row Python UDF
# @udf(returnType=StringType())
# def categorize(revenue):
#     if revenue < 100: return 'low'
#     ...

# Fast: Pandas (vectorized) UDF
@pandas_udf(StringType())
def categorize_vectorized(revenue: pd.Series) -> pd.Series:
    return revenue.apply(
        lambda r: 'low' if r < 100 else ('mid' if r < 1000 else 'high')
    )

df = df.withColumn('bucket', categorize_vectorized('revenue'))

Expect these follow-ups

• What is the difference between a scalar Pandas UDF and a grouped map Pandas UDF?
• How would you detect in the Spark UI that a Python UDF is causing a bottleneck?