Data Manipulation and Transformation Questions

Encompasses techniques and best practices for cleaning, transforming, and preparing data for analysis and production systems. Candidates should be able to handle missing values, duplicates, inconsistency resolution, normalization and denormalization, data typing and casting, and validation checks. Expect discussion of writing robust code that handles edge cases such as empty datasets and null values, defensive data validation, unit and integration testing for transformations, and strategies for performance and memory efficiency. At more senior levels include design of scalable, debuggable, and maintainable data pipelines and transformation architectures, idempotency, schema evolution, batch versus streaming trade offs, observability and monitoring, versioning and reproducibility, and tool selection such as SQL, pandas, Spark, or dedicated ETL frameworks.

MediumTechnical

57 practiced

A Spark job must process JSON logs with deeply nested structures. Provide a transformation plan to normalize nested arrays into a wide analytical table, ensuring you handle optional fields, nested nulls, and missing arrays. Show example PySpark code to explode and select fields safely.

Sample Answer

Approach:- Infer or define a nested schema for the JSON, parse it into a DataFrame.- Use explode_outer to safely unnest arrays (returns NULL when array missing/empty).- Use select with safe navigation (col("struct.field") or getField/getItem) and coalesce/defaults to handle optional fields and nested nulls.- Aggregate or pivot as needed to produce a wide analytical table.- Keep transformations idempotent and document null semantics.

Example PySpark code (PySpark):

python

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, explode_outer, coalesce, lit
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, LongType, DoubleType

spark = SparkSession.builder.getOrCreate()

# Example schema -- define key nested structures explicitly for performance & safety
schema = StructType([
    StructField("event_id", StringType()),
    StructField("user", StructType([
        StructField("id", StringType()),
        StructField("profile", StructType([
            StructField("age", LongType()),
            StructField("prefs", ArrayType(StructType([
                StructField("k", StringType()),
                StructField("v", StringType())
            ])))
        ]))
    ])),
    StructField("actions", ArrayType(StructType([
        StructField("type", StringType()),
        StructField("value", DoubleType()),
        StructField("meta", StructType([StructField("src", StringType())]))
    ])))
])

# Load raw JSON lines and parse
raw = spark.read.text("/path/to/logs/*.json")
df = raw.select(from_json(col("value"), schema).alias("j")).select("j.*")

# Explode actions safely (handles missing/empty arrays)
df_actions = df.withColumn("action", explode_outer(col("actions")))

# Explode user.profile.prefs safely (if you need one pref per row)
df_prefs = df.withColumn("pref", explode_outer(col("user.profile.prefs")))

# Build wide analytical row by selecting fields with defaults
wide = df_actions.join(df_prefs.select("event_id", "pref"), on="event_id", how="left_outer") \
    .select(
        col("event_id"),
        coalesce(col("user.id"), lit("unknown")).alias("user_id"),
        col("user.profile.age").alias("user_age"),
        col("action.type").alias("action_type"),
        coalesce(col("action.value"), lit(0.0)).alias("action_value"),
        col("action.meta.src").alias("action_src"),
        col("pref.k").alias("pref_key"),
        col("pref.v").alias("pref_value")
    )

Key points:- explode_outer returns a NULL row for missing/empty arrays so you can still preserve the parent record.- coalesce + literals enforce analytic-friendly defaults.- Defining schema avoids costly schema inference and prevents silent type surprises.- If arrays are large, consider explode + aggregation to limit row explosion (e.g., take first N, aggregate map).Complexity:- Time: O(rows + total_array_elements) — exploding multiplies rows by array sizes.- Space: proportional to output rows; ensure partitions and shuffle are tuned.

Edge cases:- Deeply missing paths → nulls; use coalesce or map defaults.- Extremely large arrays → explode explosion; consider sampling/aggregation.- Mixed schema versions → maintain schema evolution handling (sparse fields, version tags).

Alternatives:- Use posexplode_outer to keep index of each element.- Use map_from_entries to pivot key-value prefs into columns if keys are known.

MediumTechnical

65 practiced

Write a PySpark example that implements an upsert (merge) of CDC events into a Delta Lake table. Assume CDC events are in a streaming DataFrame with keys: id, op_type ('I','U','D'), payload (struct), ts. Show how you would apply changes to keep a current view table consistent and idempotent.

Sample Answer

Approach: use Structured Streaming to read CDC events, deduplicate per id by highest ts (and op priority), and apply idempotent MERGE INTO Delta table in foreachBatch. Maintain metadata columns in target: payload, _last_ts, _is_deleted. Use MERGE with conditions that only apply when incoming ts >= target._last_ts to ensure idempotency and ordering.

python

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

# incoming streaming DF schema: id, op_type ('I','U','D'), payload (struct), ts (timestamp)
cdc_stream = spark.readStream.format("kafka")...  # omitted: read and parse into columns above

# assign numeric op priority so Delete > Update > Insert if same ts (optional)
op_priority = F.when(F.col("op_type") == "D", 3).when(F.col("op_type") == "U", 2).otherwise(1)

# In each microbatch, pick the latest event per id (highest ts, then highest priority)
def upsert_batch(batch_df, batch_id):
    if batch_df.rdd.isEmpty():
        return

    dedup = (batch_df
             .withColumn("_op_prio", op_priority)
             .withColumn("_ts", F.col("ts").cast("long"))
             .withColumn("_payload", F.col("payload"))
             .select("id", "op_type", "_payload", "_ts", "_op_prio")
             .withColumn("rank", F.row_number().over(
                 Window.partitionBy("id").orderBy(F.col("_ts").desc(), F.col("_op_prio").desc())
             ))
             .filter(F.col("rank") == 1)
             .drop("rank"))

    # write dedup to temp view or temp table
    tmp_path = "/tmp/cdc_batch"
    dedup.createOrReplaceTempView("cdc_events_batch")

    delta_table_path = "/mnt/delta/current_view"
    # ensure table exists with columns: id, payload (struct), _last_ts (long), _is_deleted (boolean)
    if not DeltaTable.isDeltaTable(spark, delta_table_path):
        (dedup
         .withColumn("payload", F.col("_payload"))
         .withColumn("_last_ts", F.col("_ts"))
         .withColumn("_is_deleted", F.when(F.col("op_type") == "D", True).otherwise(False))
         .select("id","payload","_last_ts","_is_deleted")
         .write.format("delta").mode("overwrite").save(delta_table_path))

    delta = DeltaTable.forPath(spark, delta_table_path)

    src = spark.table("cdc_events_batch").alias("src")

    # Only apply if incoming ts >= target._last_ts (or target null)
    delta.alias("t").merge(
        source=src,
        condition="t.id = src.id"
    ).whenMatchedUpdate(
        condition="src._ts >= t._last_ts",
        set={
            "payload": "src._payload",
            "_last_ts": "src._ts",
            "_is_deleted": "CASE WHEN src.op_type = 'D' THEN true ELSE false END"
        }
    ).whenNotMatchedInsert(
        condition="true",
        values={
            "id": "src.id",
            "payload": "src._payload",
            "_last_ts": "src._ts",
            "_is_deleted": "CASE WHEN src.op_type = 'D' THEN true ELSE false END"
        }
    ).execute()

query = (cdc_stream
         # parse/transform as needed to columns: id, op_type, payload, ts
         .writeStream
         .foreachBatch(upsert_batch)
         .option("checkpointLocation", "/mnt/checkpoints/cdc_upsert")
         .start())

Key points:- Deduplicate per id per batch by ts and op priority to keep deterministic ordering.- MERGE condition enforces idempotency by comparing incoming ts to stored _last_ts.- Store _is_deleted flag instead of physically deleting rows (soft delete) to preserve history and simplify out-of-order handling.Edge cases:- Out-of-order events: rely on timestamps, consider watermarks and late-arrival strategy.- Exactly-once semantics: Delta + checkpointing + foreachBatch ensures atomic batch commits.- Schema evolution: ensure payload struct evolves safely (use Delta schema evolution options).

HardTechnical

77 practiced

You receive a stream of CDC events (insert/update/delete) from multiple regions. Events may arrive out-of-order. Design an algorithm to merge events into an SCD Type 2 store such that final temporal order is respected and late events are applied correctly. Discuss watermarking, buffering, and reprocessing trade-offs.

Sample Answer

Requirements & constraints:- Input: CDC events (insert/update/delete) from multiple regions, may be out-of-order, possibly delayed/duplicated.- Output: SCD Type 2 table with non-overlapping versions per business key, correct temporal order, and late-arrival correction.

High-level algorithm:1. Enrich each event with event_time (source-ts) and ingestion_time.2. Partition by business key; within each partition, order by event_time (tie-breaker: ingestion_time, region priority, or event id).3. Use a buffering + watermarking window per key to tolerate lateness, then produce ordered ops that convert to SCD2 upserts.

Implementation sketch:- Stream processor (e.g., Flink/Spark Structured Streaming) does: - Assign event_time and event-time watermarks (global or per-region). - Buffer events per key in an event-time hold (max_allowed_lateness L). - When watermark > event_time + L, sort buffered events up to watermark and emit deterministic operations: close prior active record (set end_ts = current_event_time - epsilon), insert new record with start_ts = event_time and end_ts = +inf for inserts/updates, mark deletes as end_ts = event_time.

Late-event handling / reprocessing:- For events arriving after watermark (late events), options: a) Reject/route to DLQ for manual/reprocess (low correctness). b) Support incremental correction: reapply event by generating compensating SCD2 deltas — perform read-modify-write to find affected records and rewrite ranges (requires idempotent, versioned writes). c) Full backfill/replay: reprocess historical window (expensive) using persisted raw CDC log and rewrite SCD2 partitions (best correctness).

Trade-offs:- Watermark length L: larger L => fewer late corrections but higher latency and state. Small L => low latency, more reprocessing.- Buffering per key/state size: needs limits and eviction; use spill-to-disk or external state store.- Exactly-once vs at-least-once: prefer transactional sinks (upsert with version check) to avoid duplicates.- Complexity: incremental compensations need careful concurrency control (optimistic locking or transactions) and can be complex but avoid full re-writes.

Best practice:- Persist raw CDC immutable log for replay.- Choose L based on SLA and observed delay percentiles (95/99th).- Implement late-event pipeline for automatic corrections using deterministic re-sequencing and idempotent SCD2 write pattern (write new versions with version_id and tombstones).- Monitor latency, late-rate, and state growth; alert and tune L or trigger periodic backfills.

MediumSystem Design

68 practiced

You're tasked with building an incremental (daily) ingestion pipeline from a transactional OLTP Postgres DB into a data warehouse. Describe two approaches: (1) time-window incremental pull and (2) CDC-based ingestion. For each, list implementation steps, pros/cons, and how you'd ensure correctness in presence of late updates/deletes.

Sample Answer

Requirements & constraints (assumed): daily/incremental from Postgres OLTP → DW; need correctness (no lost updates/deletes), reasonable latency (daily), bounded cost.

1) Time-window incremental pullApproach summary: once-per-day extract rows changed since last successful run using a timestamp or high-watermark column.

Implementation steps:- Add/confirm a reliable change column: last_modified_ts (or use updated_at trigger) and a monotonically increasing primary key.- Pipeline: record last_successful_watermark (timestamp or max id) → query source: SELECT * FROM table WHERE last_modified_ts > watermark AND last_modified_ts <= now_cutoff → extract → transform → load into DW via upsert (merge) or SCD logic → persist new watermark (max last_modified_ts processed).- Add nightly snapshot job for new tables + schema drift detection.

Pros:- Simple to implement with SQL; low operational complexity.- Works without DB-level replication access.- Easier to reason about cost for daily batch.

Cons:- Relies on correct last_modified_ts; clock skew, missed triggers or bulk updates that don't change timestamp can lose changes.- Hard to capture deletes unless soft-delete flag exists or you scan for missing PKs.- Late-arriving updates whose last_modified_ts is older than the watermark are missed.

Ensuring correctness / handling late updates & deletes:- Use safety window / lookback: when querying use watermark minus delta (e.g., reprocess last 24 hours) to re-check recently processed rows, and deduplicate by primary key during load (idempotent upserts).- Implement tombstone records: if deletes are recorded via soft_delete flag or an audit table, include them.- Periodic reconciliation: daily or weekly full-key-count checksums or sampling; run full/partial re-scan for tables with high importance.- Use transactional consistency: snapshot the source at query time (SELECT ... FOR SHARE? or use read-after-write guarantees) or run extraction during low-load with pg_dump snapshot or serializable snapshot.

2) CDC-based ingestionApproach summary: capture every commit-level change (insert/update/delete) from Postgres WAL to stream changes continuously; apply to DW as upserts and deletes.

Implementation steps:- Enable logical replication on Postgres; configure replication slot.- Use CDC tool: Debezium or commercial CDC to read WAL and output change events (format: before/after, op type, txid, lsn, commit_ts).- Stream to Kafka or cloud pubsub; consumer transforms events and writes to DW using exactly-once or idempotent upsert/delete semantics (MERGE in Snowflake/BigQuery/Redshift).- Maintain offsets (LSN) checkpointing in durable store; handle schema changes via schema registry.- Backfill: initial snapshot + apply CDC stream from snapshot LSN.

Pros:- Fine-grained, complete capture of inserts/updates/deletes; low risk of missed changes.- Efficient for large tables; only changes transmitted.- Supports near-real-time replication if desired.

Cons:- More operational complexity: replication slots, tooling, schema evolution, Kafka management.- Requires DB privileges and possible tuning (WAL retention).- More components to monitor and maintain.

Ensuring correctness / handling late updates & deletes:- Use initial consistent snapshot: tools create a consistent snapshot and then apply changes from WAL to avoid gaps.- Use LSN / txid ordering: consumer applies events in commit order using LSN/txid to ensure correct sequence; include idempotent application (use upsert keyed by PK and last_applied_lsn or commit_ts).- Handle duplicates and retries: include unique event ids and de-dup logic; use MERGE or write-side dedup tables.- Maintain tombstones for deletes and ensure downstream respects delete events (hard delete vs soft-delete policies).- Schema evolution: use schema registry and migration strategy to avoid data loss.- Monitor replication lag and WAL retention; alert if slot retention risks WAL removal.

When to choose:- Use time-window pull for small teams, simple tables, limited access, low change volumes, and when daily latency is acceptable.- Use CDC when correctness for deletes/updates is critical, data volume/velocity is higher, or downstream requires near-real-time.

MediumSystem Design

73 practiced

Design a data validation framework for nightly ETL jobs that: (1) checks row counts, nullness, value ranges, distribution changes vs baseline, and referential integrity; (2) surfaces failures to Slack and creates Jira tickets when thresholds are exceeded; (3) allows rule versioning and dry-run mode. Provide high-level components, data flow, and example validation rule DSL.

Sample Answer

Requirements clarified:- Run nightly after ETL, validate row counts, nullness, value ranges, distribution drift vs baseline, referential integrity- Alert to Slack and auto-create Jira when thresholds exceeded- Support rule versioning and dry-run mode

High-level components:1. Validation Rules Store: Git-backed repository (YAML/JSON) + metadata DB for versions.2. Rule Compiler: Parses DSL -> executable checks (SQL/Spark queries).3. Validator Engine: Runs checks against datasets (Spark jobs or SQL), computes metrics, diffs vs baseline.4. Baseline Store: Historical metrics (row counts, histograms, null rates) in time-series DB (e.g., Prometheus/Influx) or Parquet on object store.5. Orchestrator: Airflow job triggers ETL then Validator; supports dry-run flag.6. Alerting & Ticketing Service: Applies thresholds, posts to Slack, creates Jira via API when severity > threshold.7. UI / Dashboard: Shows rule versions, recent runs, failures, and enables rule rollback.

Data flow (stepwise):1. ETL completes → Airflow marks success and triggers Validator with dataset location and run_id.2. Rule Compiler loads current rule version (or specific version), compiles to queries.3. Validator Engine executes checks (Spark jobs) producing metrics per rule and run.4. Compare metrics to Baseline Store; compute drift (e.g., KL divergence, PSI).5. Store run results and metrics; evaluate thresholds. In dry-run mode: record results but do not call Alerting.6. If thresholds exceeded, Alerting posts to Slack channel with contextual links and creates Jira with template (dataset, run_id, failing rules, sample rows).7. Dashboard shows status and supports re-run or rule edits (creates new rule version in Rules Store).

Example validation rule DSL (YAML):

yaml

version: 2
rules:
  - id: orders_rowcount
    description: "Row count shouldn't drop >10% vs baseline"
    type: row_count
    dataset: prod.orders
    comparator: pct_change
    threshold: -0.10
    severity: high

  - id: orders_total_nulls
    description: "order_total should have <1% nulls"
    type: null_rate
    dataset: prod.orders
    column: order_total
    operator: "<="
    threshold: 0.01
    severity: medium

  - id: order_amount_range
    description: "order_amount within expected range"
    type: value_range
    dataset: prod.orders
    column: order_amount
    min: 0.0
    max: 10000.0
    severity: high

  - id: order_amount_dist
    description: "Distribution drift vs baseline (PSI)"
    type: histogram_drift
    dataset: prod.orders
    column: order_amount
    bins: 20
    metric: psi
    threshold: 0.10
    severity: medium

  - id: fk_orders_customers
    description: "orders.customer_id must reference customers.id"
    type: referential_integrity
    left: prod.orders.customer_id
    right: prod.customers.id
    operator: >=
    threshold: 0.995   # allowed fraction of valid refs
    severity: high

Key design notes and trade-offs:- Use Spark for scale; for small tables use direct SQL.- Baselines: rolling-aggregates (7/30-day) reduce noise; store raw sample snapshots for debugging.- Dry-run: enforce by passing flag to Orchestrator/Alerting so results are recorded but no external notifications are sent.- Versioning: keep rules in Git; metadata DB maps rule set versions to runs for audit and reproducibility.- Alert throttling & deduplication to avoid noise; include auto-snooze if Jira exists.- Security: Jira/Slack tokens in secrets manager; RBAC for rule edits.

This design balances scalability, reproducibility, and fast troubleshooting for nightly ETL validation.

Unlock Full Question Bank

Get access to hundreds of Data Manipulation and Transformation interview questions and detailed answers.

Join thousands of developers preparing for their dream job.