Real Time and Batch Ingestion Questions

Focuses on choosing between batch ingestion and real time streaming for moving data from sources to storage and downstream systems. Topics include latency and throughput requirements, cost and operational complexity, consistency and delivery semantics such as at least once and exactly once, idempotent and deduplication strategies, schema evolution, connector and source considerations, backpressure and buffering, checkpointing and state management, and tooling choices for streaming and batch. Candidates should be able to design hybrid architectures that combine streaming for low latency needs with batch pipelines for large backfills or heavy aggregations and explain operational trade offs such as monitoring, scaling, failure recovery, and debugging.

MediumTechnical

131 practiced

Using Apache Spark Structured Streaming (PySpark), describe (pseudocode is fine) how you would compute a 5-minute tumbling window aggregation of events grouped by user_id, handling late events up to 10 minutes with watermarks, and write results as partitioned Parquet files to S3. Mention checkpointing and write semantics.

Sample Answer

Approach summary: use event-time 5-minute tumbling windows with a 10-minute watermark so late events within 10 minutes are included; aggregate per user_id; write Parquet files partitioned by date to S3 using checkpointing and append semantics. Ensure S3-commit compatibility (S3A/EMR committers) for atomic writes and enable a stable checkpointLocation for fault tolerance.

python

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col, to_timestamp, date_format

spark = SparkSession.builder.appName("user-window-agg").getOrCreate()

# Read streaming source (e.g., Kafka)
raw = (spark.readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "kafka:9092")
       .option("subscribe", "events")
       .load())

events = (raw.selectExpr("CAST(value AS STRING) as json")
          .selectExpr("from_json(json, 'user_id STRING, event_ts STRING, value DOUBLE') as v")
          .selectExpr("v.user_id", "to_timestamp(v.event_ts) as event_time", "v.value"))

# Windowed aggregation with 5-minute tumbling windows and 10-minute watermark
agg = (events
       .withWatermark("event_time", "10 minutes")                # drop events later than watermark
       .groupBy(col("user_id"),
                window(col("event_time"), "5 minutes"))          # tumbling window
       .agg({"value": "sum", "*": "count"})
       .withColumn("window_start", col("window").start)
       .withColumn("window_end", col("window").end)
       .withColumn("event_date", date_format(col("window_start"), "yyyy-MM-dd")))

# Write to S3 as partitioned Parquet with checkpointing
(out := agg.writeStream
       .format("parquet")
       .outputMode("append")                                      # append is appropriate with watermark
       .option("path", "s3a://my-bucket/path/to/output/")
       .option("checkpointLocation", "s3a://my-bucket/checkpoints/user-window-agg/")
       .partitionBy("event_date")                                 # partition files by date
       .trigger(processingTime="1 minute")
       .start())

Key points / reasoning:- withWatermark("event_time","10 minutes") allows late events up to 10 minutes; events later are ignored.- Tumbling window of 5 minutes via window(...,"5 minutes") groups event_time into fixed 5-min buckets.- outputMode "append" works because watermark + windowed aggregation lets Spark emit results once windows are complete; using "complete" would rewrite full state each trigger and isn't supported for file sinks.- CheckpointLocation is required for exactly-once recovery of streaming job state; for S3, use S3A/EMR committers or a durable filesystem to ensure atomic file commits.- Monitor state size (late threshold) and set spark.sql.streaming.stateStore.maintenanceInterval / state retention as needed.

EasyTechnical

91 practiced

Define 'at least once', 'at most once', and 'exactly once' delivery semantics in the context of stream ingestion. For each semantic give a concrete example of a failure mode that can lead to duplicate or missing data, and one practical mitigation technique.

MediumTechnical

73 practiced

Implement idempotent write logic in pseudocode for a streaming job that writes user profile events to a database which only supports upserts by key. Describe how you would avoid race conditions from concurrent writers and ensure correctness under retries.

Sample Answer

To implement idempotent writes for streaming user-profile events with only key-based upserts, use a write model that includes a durable, monotonic version (or event_id/timestamp) per record and perform conditional upserts (compare-and-set) to avoid races. Use optimistic concurrency with a conditional WHERE or DB-specific upsert that checks version, and fall back to retry with backoff.

``python

# pseudocode (python-style)
def process_event(event):
    key = event.user_id
    incoming_version = event.version  # monotonic increasing (e.g., event_sequence or timestamp)
    payload = event.payload

    # Try conditional upsert: only apply if incoming_version > stored_version
    # Implementation depends on DB; shown as single atomic statement
    success = db.execute(
      """
      INSERT INTO profiles (user_id, version, data)
      VALUES (:key, :v, :data)
      ON CONFLICT (user_id) DO UPDATE
      SET version = EXCLUDED.version, data = EXCLUDED.data
      WHERE profiles.version < EXCLUDED.version
      """,
      params={"key": key, "v": incoming_version, "data": payload}
    )

    if not success:
        # Another writer had equal/greater version. Verify stored row to decide:
        stored = db.query("SELECT version, data FROM profiles WHERE user_id = :key", {"key":key})
        if stored.version == incoming_version and stored.data == payload:
            return "idempotent - no-op"
        else:
            # if incoming_version <= stored.version, drop; if equal but different payload, resolve per policy
            resolve_conflict(event, stored)
    return "written"

Key points:- Use a monotonically increasing version (event sequence or logical timestamp) as authoritative.- Perform single-statement conditional upsert (atomic) to avoid races.- Handle retries by treating the operation idempotent: repeated same-version writes either succeed once or are no-ops.- If DB lacks conditional upsert, use lightweight transactions / IF / CAS (e.g., SELECT ... FOR UPDATE then UPDATE) but beware reduced concurrency.- Add exponential backoff and max retries for transient conflicts.

Complexity & trade-offs:- Single conditional upsert is O(1) per event; contention may cause retries.- Using DB transactions/FOR UPDATE increases latency and reduces throughput but guarantees serializability.- If events can arrive out-of-order, use vector clocks or accept last-writer-wins only if timestamps are reliable.

Edge cases:- Missing/duplicate version fields: reject or derive using event_id.- Equal version with different payloads: decide deterministic tie-breaker (e.g., compare producer_id).- Schema migrations: ensure version field always present.

HardTechnical

77 practiced

A large reprocessing job produced duplicated records and skewed aggregates in your analytics data lake. Propose a step-by-step remediation plan to repair historical data with minimal downtime, ensure reproducibility, and provide an audit trail of corrections.

Sample Answer

Clarify scope & constraints- Confirm which tables/datasets and time ranges are affected, SLA for downtime, storage/backups available, and who must be notified.

Plan (step-by-step)1. Freeze writes or route new input to a staging stream - Short freeze of downstream consumers or switch producers to write to an isolated “staging” topic/bucket to prevent further contamination.

2. Snapshot current state (immutable) - Create read-only snapshots / immutable copies of affected dataset partitions (e.g., S3 prefix copy, HDFS snapshot, or Delta/Hudi table version). Tag them with job-id/timestamp for audit.

3. Detect & mark bad records - Run detection job (Spark) to identify duplicates/skew using deterministic keys, checksums, timestamps. Output a “corrections” manifest describing record ids, partitions, root cause, and suggested action. Store manifest in versioned metadata store (e.g., Git, S3 with versioning).

4. Construct idempotent repair transformations - Implement deterministic dedupe/aggregate recompute logic (groupBy key + window, stable sort by event_time, apply dedupe rules). Package as parameterized Spark job that reads snapshot + manifest and produces repaired partitions. Ensure job is idempotent (re-running yields same result).

5. Run test repairs on a sample - Execute on a small partition set; validate counts/aggregates vs expected; produce data quality report and diff (old vs new) stored alongside manifests.

6. Audit & approval - Produce audit bundle: snapshots, manifest, test diffs, validation metrics. Require sign-off from data owners before bulk apply.

7. Bulk repair with minimal downtime - Use partition-level atomic swaps: write repaired partitions to a staging location and then perform atomic metadata update (e.g., Delta Lake transaction commit, Hudi commit, or S3 manifest swap). This allows consumers to keep reading unaffected partitions until swap time, minimizing downtime.

8. Post-repair validation & monitoring - Run full validation suite: row counts, aggregate comparisons, checksums. Publish audit logs and notify stakeholders. Monitor downstream dashboards for anomalies.

9. Rollback plan - Keep snapshots and manifests to revert via metadata rollback (Delta/Hudi allow time-travel) if issues occur.

Operational & reproducibility practices- Store all code, configs, manifests, and run parameters in version control; tag job runs with unique IDs; log provenance (input snapshot id, job id, operator, timestamp).- Emit structured audit entries for each corrected partition: reason, original snapshot pointer, corrected snapshot pointer, checksums.- Automate with CI/CD pipelines and parameterized airflow/kubernetes jobs to enable repeatable runs.

Example technologies- Spark for processing, Delta Lake/Hudi/ Iceberg for atomic commits and time-travel, S3/GCS with versioning for snapshots, Airflow for orchestration, and an audit store (Postgres or Elastic) for manifests/logs.

Key trade-offs- Full-table rewrites are safest but expensive; prefer partition-level repair when possible. Maintainability vs speed: favor reproducibility and auditability even if repair takes longer.

This plan minimizes downtime by isolating new writes, uses atomic swaps to switch repaired partitions, ensures reproducibility via snapshots/versioned manifests and idempotent jobs, and provides a complete audit trail for compliance and rollback.

HardTechnical

81 practiced

Implement pseudocode for a sliding-window deduplication that drops duplicates of event_id within a 1-hour sliding window using limited memory. Describe data structures and expiration logic. Assume high ingest rate and requirement to evict old keys efficiently.

Sample Answer

Approach: Use a combination of a time-partitioned hash table (buckets per small time slot) + an in-memory LRU-like index to evict keys efficiently. Store only event_id -> last-seen-timestamp and maintain buckets of event_ids by slot (e.g., 1-minute slots) so eviction of everything older than 1 hour is O(#expired). This bounds memory and avoids scanning entire map.

python

# Pseudocode (python-like)
SLOT_SECONDS = 60            # 1-minute slots
WINDOW_SECONDS = 3600        # 1 hour
NUM_SLOTS = WINDOW_SECONDS // SLOT_SECONDS  # 60

# Data structures
last_seen = {}               # dict: event_id -> timestamp (for quick existence check)
slots = [set() for _ in range(NUM_SLOTS)]  # circular buffer of sets of event_ids

def slot_index(ts):
    return (ts // SLOT_SECONDS) % NUM_SLOTS

current_head_ts = floor_now_to_slot(now())  # start timestamp of current slot

def ingest(event_id, ts):
    global current_head_ts
    advance_slots_if_needed(ts)

    # Drop duplicate if seen within window
    if event_id in last_seen and ts - last_seen[event_id] < WINDOW_SECONDS:
        return False  # duplicate, drop

    # Accept event: record last_seen and add to current slot
    last_seen[event_id] = ts
    slots[slot_index(ts)].add(event_id)
    return True  # accepted

def advance_slots_if_needed(ts):
    global current_head_ts
    target_head = floor_to_slot(ts)
    while current_head_ts < target_head:
        # expire the oldest slot
        idx = slot_index(current_head_ts)
        for eid in slots[idx]:
            # remove only if last_seen matches the slot's time (handles re-inserts)
            if last_seen.get(eid, 0) <= current_head_ts + SLOT_SECONDS - 1:
                del last_seen[eid]
        slots[idx].clear()
        current_head_ts += SLOT_SECONDS

Key points:- Per-event O(1) checks and inserts.- Eviction cost proportional to number of events in expired slot(s), not total keys.- Use small slot size (e.g., 1s–60s) to trade memory vs eviction granularity.- For high ingest: shard structures across workers (by hash(event_id)) and persist occasional snapshots if needed.Edge cases:- Out-of-order events: compare ts with last_seen; accept if outside window.- Clock skew: prefer ingest-time watermarking or use event-time with lateness allowance.Alternatives:- Bloom filter per slot for memory savings (probabilistic, requires periodic resets).- Count-min sketch variants for approximate dedupe.

Unlock Full Question Bank

Get access to hundreds of Real Time and Batch Ingestion interview questions and detailed answers.

Join thousands of developers preparing for their dream job.