Data platform engineer coding interview questions

As asked

Write a PySpark snippet that reads a staging DataFrame with potential duplicates on (user_id, event_id), deduplicates keeping the latest record by event_ts, then merges into an Iceberg target table updating existing rows and inserting new ones.

Sample answer outline

The candidate should use Window.partitionBy('event_id').orderBy(desc('event_ts')) with row_number() to deduplicate, filter to row_number == 1, then use the Iceberg MERGE INTO SQL extension (registered via the Iceberg Spark extensions config) or the DataFrameWriterV2 overwritePartitions API. They should handle matched (update) and not-matched (insert) clauses correctly. Note: DeltaTable is a Delta Lake class and has no equivalent in Iceberg; the correct Iceberg merge entry point is the MERGE INTO SQL statement or the V2 write API.

Reference implementation (python)

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

w = Window.partitionBy("event_id").orderBy(F.desc("event_ts"))
deduped = (
    staging
    .withColumn("rn", F.row_number().over(w))
    .filter(F.col("rn") == 1)
    .drop("rn")
)

deduped.createOrReplaceTempView("staged")
spark.sql("""
  MERGE INTO prod.events t
  USING staged s ON t.event_id = s.event_id
  WHEN MATCHED THEN UPDATE SET *
  WHEN NOT MATCHED THEN INSERT *
""")

Expect these follow-ups

• How would you handle the case where event_ts can be identical for two records with the same event_id?
• What is the performance difference between using MERGE INTO SQL and the Iceberg DataFrameWriterV2 overwritePartitions?

As asked

Write a Java snippet that uses the Iceberg RewriteDataFiles action to compact files smaller than 64 MB into target files of 512 MB in the 'events' table, using bin-packing strategy.

Sample answer outline

The candidate should instantiate a Catalog, load the table, and call Actions.forTable(table).rewriteDataFiles() with the appropriate options for min-file-size-bytes, target-file-size-bytes, and strategy set to BIN_PACK. They should call execute() and handle the RewriteDataFilesResult. Bonus for wrapping in a try-with-resources and committing only on success.

Reference implementation (java)

Table table = catalog.loadTable(TableIdentifier.of("db", "events"));
RewriteDataFiles.Result result = Actions.forTable(table)
    .rewriteDataFiles()
    .option(RewriteDataFiles.TARGET_FILE_SIZE_BYTES, Long.toString(512 * 1024 * 1024))
    .option(RewriteDataFiles.MIN_FILE_SIZE_BYTES, Long.toString(64 * 1024 * 1024))
    .strategy(BinPackStrategy.name())
    .execute();
System.out.printf("Rewrote %d files into %d files%n",
    result.rewrittenDataFilesCount(), result.addedDataFilesCount());

Expect these follow-ups

• How would you parallelize this across many tables using a Spark job driver?
• How do you make this action safe to run concurrently with an active Flink ingestion job?

As asked

Write a Flink DataStream API snippet in Java that reads JSON events from a Kafka topic 'raw_events', parses them into a Row, and writes them to an Iceberg table using the FlinkSink builder with exactly-once enabled.

Sample answer outline

The candidate should create a StreamExecutionEnvironment, configure KafkaSource with a JSON deserializer, apply a map to parse JSON to Row matching the Iceberg schema, then use FlinkSink.forRow(table, tableSchema).tableLoader(tableLoader).upsert(false).build() and add it as a sink. They should set the checkpoint interval and enable exactly-once checkpointing mode.

Reference implementation (java)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60_000, CheckpointingMode.EXACTLY_ONCE);

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("broker:9092")
    .setTopics("raw_events")
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<Row> rows = env
    .fromSource(source, WatermarkStrategy.noWatermarks(), "kafka")
    .map(json -> parseJson(json));  // returns Row

FlinkSink.forRow(rows, tableSchema)
    .tableLoader(tableLoader)
    .distributionMode(DistributionMode.HASH)
    .build();

env.execute("kafka-to-iceberg");

Expect these follow-ups

• How does FlinkSink handle the case where Iceberg commit fails after the Flink checkpoint succeeds?
• What changes if you need to upsert based on a primary key?

As asked

Write an Airflow DAG snippet that uses S3KeySensor to wait for the presence of a date-partitioned S3 prefix before triggering a downstream Spark job. The partition path is s3://data-lake/events/dt={{ ds }}/. The sensor should time out after 2 hours and not block a DAG slot while waiting.

Sample answer outline

The candidate should use S3KeySensor with bucket_name and bucket_key using template variables, set timeout=7200, poke_interval=300, and mode='reschedule' so the sensor releases its worker slot between pokes. They should wire it as an upstream dependency of the Spark operator.

Reference implementation (python)

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

wait_for_partition = S3KeySensor(
    task_id="wait_for_events_partition",
    bucket_name="data-lake",
    bucket_key="events/dt={{ ds }}/",
    wildcard_match=False,
    timeout=7200,
    poke_interval=300,
    mode="reschedule",
    dag=dag,
)

run_spark = SparkSubmitOperator(
    task_id="process_events",
    application="s3://jobs/process_events.py",
    dag=dag,
)

wait_for_partition >> run_spark

Expect these follow-ups

• How does mode='reschedule' affect the Airflow worker pool compared to mode='poke'?
• What would you do if the upstream data is delayed by more than 2 hours occasionally?

As asked

Write a PySpark snippet that reads the Change Data Feed from a Delta table starting from version 5, filters for insert and update operations only, and writes the results to a downstream Iceberg table.

Sample answer outline

The candidate should use spark.read.format('delta').option('readChangeFeed', 'true').option('startingVersion', 5).load('path/to/table'), then filter where _change_type is in ('insert', 'update_postimage'), drop the CDF metadata columns, and write to the Iceberg target using the V2 writer. They should discuss tracking the latest processed version for the next incremental run.

Reference implementation (python)

cdf = (
    spark.read.format("delta")
    .option("readChangeFeed", "true")
    .option("startingVersion", 5)
    .load("s3://lake/customers")
)

filtered = cdf.filter(
    cdf["_change_type"].isin("insert", "update_postimage")
).drop("_change_type", "_commit_version", "_commit_timestamp")

filtered.writeTo("catalog.prod.customers_iceberg").append()

Expect these follow-ups

• How do you persist the last processed version so the next job run picks up where this one left off?
• What does update_preimage versus update_postimage contain, and when would you need both?

As asked

Write a small Python function that embeds documents in batches, retries transient failures, and preserves the mapping from document id to embedding.

Sample answer outline

The implementation should batch by count or token budget, keep ids alongside text, and return a dictionary keyed by document id so downstream indexing cannot mix rows. Retries should be bounded and should only catch transient provider errors in real production code. A robust answer records failed ids for later replay rather than silently dropping them. The interviewer is looking for ordinary production hygiene around model APIs: rate limits, idempotency, logging, and partial failure. Candidates often assume the provider response order cannot surprise them and never validate lengths.

Reference implementation (python)

from collections.abc import Callable
from time import sleep

EmbedFn = Callable[[list[str]], list[list[float]]]

def embed_documents(
    docs: list[tuple[str, str]],
    embed: EmbedFn,
    batch_size: int = 64,
    max_retries: int = 3,
) -> dict[str, list[float]]:
    results: dict[str, list[float]] = {}

    for start in range(0, len(docs), batch_size):
        batch = docs[start : start + batch_size]
        ids = [doc_id for doc_id, _ in batch]
        texts = [text for _, text in batch]

        for attempt in range(max_retries):
            try:
                vectors = embed(texts)
                if len(vectors) != len(ids):
                    raise ValueError("embedding response length mismatch")
                results.update(dict(zip(ids, vectors)))
                break
            except TimeoutError:
                if attempt == max_retries - 1:
                    raise
                sleep(2 ** attempt)

    return results

Expect these follow-ups

• How would you respect provider rate limits across multiple workers?
• Where do you store failed ids for replay?
• How do you know an embedding model upgrade did not hurt retrieval?

As asked

Write a Kafka Streams Java snippet that counts page view events per user in session windows with a 10-minute inactivity gap, then outputs the session count to an output topic.

Sample answer outline

The candidate should use StreamsBuilder to consume a KStream keyed by user_id, apply groupByKey(), windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(10))), count(), and pipe the resulting KTable to an output topic with a windowed serde. They should handle the Windowed key serialization.

Reference implementation (java)

StreamsBuilder builder = new StreamsBuilder();
KStream<String, PageView> views = builder.stream("page-views",
    Consumed.with(Serdes.String(), pageViewSerde));

views
    .groupByKey()
    .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(10)))
    .count(Materialized.as("session-counts"))
    .toStream()
    .map((wk, count) -> KeyValue.pair(
        wk.key() + "@" + wk.window().start(), count))
    .to("session-count-output", Produced.with(Serdes.String(), Serdes.Long()));

Expect these follow-ups

• How does Kafka Streams handle a late record that falls within an already-closed session window?
• What is the impact on the changelog topic when session windows merge?

As asked

Write a Python snippet using Great Expectations to create a custom expectation that validates referential integrity: every value in the 'customer_id' column of the orders table must exist in the customers table's 'id' column. Assume both tables are available as pandas DataFrames.

Sample answer outline

The candidate should subclass ColumnMapExpectation or use expect_column_values_to_be_in_set with the valid customer IDs extracted from the customers DataFrame. A more robust answer defines a custom expectation class with a condition_metric that checks set membership. They should handle nulls and document the expectation metadata.

Reference implementation (python)

import great_expectations as ge

orders_gdf = ge.from_pandas(orders_df)
customers_gdf = ge.from_pandas(customers_df)

valid_customer_ids = set(
    customers_gdf["id"].dropna().astype(str).tolist()
)

result = orders_gdf.expect_column_values_to_be_in_set(
    column="customer_id",
    value_set=valid_customer_ids,
    mostly=1.0,
    result_format="COMPLETE",
)

assert result["success"], (
    f"Referential integrity failed: "
    f"{result['result']['unexpected_count']} orphaned customer_ids"
)

Expect these follow-ups

• How would you run this expectation as a checkpoint in an Airflow DAG and block the pipeline on failure?
• What is the performance concern when the customers table has 100 million rows?