Data Ingestion and Source Systems Questions

Understand different data sources and ingestion methods: batch processing (scheduled jobs), streaming (Kafka, Kinesis), API polling, database replication, and log aggregation. Know the trade-offs between real-time and batch ingestion. Understand concepts like exactly-once delivery and idempotency.

HardTechnical

37 practiced

Compare snapshot isolation, read-committed, and serializable isolation levels from the perspective of CDC and downstream ingestion guarantees. Provide examples where a CDC pipeline might observe anomalies under weaker isolation and how that affects the analytics layer.

Sample Answer

Start with short definitions and the CDC implication, then concrete anomalies and mitigations.

Definitions & CDC guarantees- Read Committed: a reader never sees uncommitted writes, but concurrent transactions’ commits can be observed incrementally. For CDC that reads WAL/logs or uses snapshotting without transaction boundaries, events may be emitted in the order commits become visible, not as a single atomic transaction. No global consistency across multiple rows.- Snapshot Isolation (SI): a transaction reads from a consistent snapshot taken at start time; writers commit atomically but SI still allows write-skew anomalies. CDC that captures a consistent snapshot (e.g., snapshot + stream of later changes) can provide per-transaction atomic visibility if it emits whole-transaction boundaries.- Serializable: strongest — history is equivalent to some serial execution. CDC that preserves serializable order (and transaction boundaries) gives downstream analytics a single consistent timeline.

Concrete anomalies a CDC pipeline may observe under weaker isolation1. Out-of-order partial transactions (Read Committed): imagine a bank transfer implemented as debit row update then credit row update in one transaction. If CDC streams per-row commits as they appear (or uses log shipping without grouping by XID), downstream may see the debit before the credit, producing transient negative balance spikes in analytic dashboards or alerting systems.2. Lost-update / overwrites (Read Committed + non-atomic readers): two concurrent updates to the same entity may be emitted in an order that overwrites an intermediate state; downstream might compute aggregates using intermediate state and then not reconcile.3. Write-skew under SI: two concurrent transactions T1 and T2 each check inventory and decrement different SKUs but violate a cross-row invariant (sum inventory < 0). If CDC emits both commits separately (even if each sees a consistent snapshot), the final state may violate business invariants; analytic checks that rely on invariant enforcement will be wrong.4. Phantom/partial visibility in sliding-window analytics: incremental aggregations built from CDC events may include rows that are later rolled back/compensated if CDC doesn’t respect transaction commit order or replay of aborted transactions — causing spikes and then compensating corrections.

Downstream effects on analytics- Incorrect aggregates: sums, counts and ratios computed over streaming events can be temporarily or permanently wrong.- Alert noise: transient anomalies (e.g., negative balances) trigger false alarms.- Wrong joins/time-travel: analytical joins expecting row-level atomicity produce wrong results if constituent rows from one transaction arrive separately.- Data lineage and reproducibility issues: inability to map analytic state to a single source-of-truth transaction complicates debugging and recomputation.

Mitigations and best practices- Capture transaction metadata: emit XID/commit LSN/commit timestamp so downstream can group and apply events atomically.- Emit transaction boundaries and apply in-order by commit timestamp or LSN; buffer until transaction commit is observed to avoid partial visibility.- Use consistent snapshots for backfills/reconciliations so initial loads and changestreams share the same view.- Make downstream idempotent and support reorder-resilient operations (upserts with versioning, tombstones).- For strict correctness (financial, inventory), prefer serializable source transactions or enforce compensating transactions/validation downstream; use reprocessing with snapshot anchors for historical correctness.- Monitor and alert for retractions/compensations frequency as a signal of isolation-related anomalies.

Trade-offsStronger isolation and full transactional CDC increase latency and complexity (buffering, larger memory), but they reduce analytic anomalies. Choose based on business SLAs: eventual correctness with compensations for high-throughput pipelines, or transactional guarantees for critical financial metrics.

HardTechnical

42 practiced

You inherited an ingestion pipeline that shows occasional latency spikes and warehouse lag during peak hours. Walk through a troubleshooting plan: which metrics and logs to examine (producer-side, broker-side, consumer-side, network, storage), likely root causes (GC, disk saturation, consumer GC, batching), and short-term and long-term mitigations.

Sample Answer

High-level approach: triage by measuring where latency / lag originates, correlate across producer → broker → consumer → network → storage, then apply short-term mitigations to restore SLAs and longer-term fixes to prevent recurrence.

Metrics & logs to examine- Producer-side: publish rate (msg/s), batch sizes, linger.ms, retries, request latency histogram, client GC logs, thread pool saturation, error/retry counts.- Broker-side (Kafka/Rabbit/etc.): ingress throughput, ISR/rebalance events, partition leader CPU, disk I/O wait, log flush latency, controller logs, JVM GC pauses, under-replicated partitions, network interface metrics.- Consumer-side (warehouse loader / Spark streaming): consumer lag per partition, poll latency, processing time per record/batch, commit latency, executor GC logs, backpressure metrics, batch sizes, checkpoint durations.- Network: packet loss, retransmits, interface errors, bandwidth saturation, cross-AZ latency.- Storage: disk utilization, IOPS, latency (read/write ms), queue depth, filesystem errors, cloud storage throttling (S3/GCS request rates).

Likely root causes (and evidence)- Producer GC or small heap pauses → bursts in publish latency and increased retries.- Broker disk saturation or fsync latency → high log flush times, growing produce request queues.- Consumer GC or slow downstream writes (warehouse) → increasing consumer lag, long processing times, large commit durations.- Batching misconfiguration (too large or too small) → spikes when batches flush (linger/size).- Network congestion or cross-region hops → correlated network latency spikes.- Sudden rebalances or leader moves → transient lag spikes and log warnings.

Short-term mitigations (restore throughput/lag)- Increase consumer parallelism or add temporary consumers to drain backlog.- Tune producer to reduce linger/batch size and increase retries/backoff to smooth bursts.- Throttle upstream producers if possible (rate-limit) to avoid overwhelming brokers.- Move consumers to faster instances or scale out Spark executors; increase commit intervals carefully.- Temporarily increase broker resources (CPU/disk IOPS), move hot partitions, or enable faster disks.- Apply JVM GC tuning (use G1/ZGC for low-pause), increase heap to avoid full GCs short-term.

Long-term mitigations (prevent recurrence)- Right-size batching: pick producer.batch.size and linger.ms to balance latency vs throughput; enable compression to reduce IO.- Capacity plan for broker storage IOPS and network; use SSDs, provisioned IOPS, or dedicated throughput.- Implement autoscaling for consumers and backpressure-aware producers; circuit breaker to shed load.- Improve observability: end-to-end tracing, per-partition lag alerts, GC/OS-level dashboards, and anomaly detection.- Use partitioning strategy to spread load; reassign hot partitions proactively.- Harden consumer idempotency and transactional writes to warehouse; optimize downstream sinks (bulk/batched writes).- Regular chaos tests and load tests to validate behavior under peak and failure scenarios.

When presenting findings, include cross-correlation graphs (produce latency vs broker disk latency vs consumer lag) and a prioritized action plan: immediate triage steps, 24–72h mitigations, and roadmap for structural fixes.

HardTechnical

43 practiced

Given a Kafka topic of user events where duplicates may exist and you must produce sessionized outputs, write a Spark SQL query (or outline the Spark Structured Streaming logic) that deduplicates by event_id, assigns events to sessions per user with a 30-minute inactivity gap, and outputs session start, end, and event count. Include a small sample input and expected output in your explanation.

Sample Answer

Approach: use Structured Streaming with event-time watermarks to deduplicate by event_id, then sessionize per user with 30-minute inactivity gap using session_window, and aggregate session start, end, and event count. Deduplication must happen after watermarking to bound state.

Spark Structured Streaming (PySpark) logic:

python

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr, col

spark = SparkSession.builder.appName("sessions").getOrCreate()

# Read Kafka (assume value is JSON with event_id, user_id, event_time)
raw = spark.readStream.format("kafka") \
  .option("kafka.bootstrap.servers","broker:9092") \
  .option("subscribe","user_events") \
  .load()

events = raw.selectExpr("CAST(value AS STRING) as json") \
  .selectExpr("from_json(json, 'event_id STRING, user_id STRING, event_time TIMESTAMP') as data") \
  .select("data.*")

# Apply watermark, deduplicate by event_id within watermark window, sessionize with 30m gap
deduped = events.withWatermark("event_time", "1 hour") \
    .dropDuplicates(["event_id"])

sessions = deduped.groupByExpr("user_id", "session_window(event_time, '30 minutes') as sess") \
    .agg(
      expr("min(sess.start) as session_start"),
      expr("max(sess.end) as session_end"),
      expr("count(*) as event_count")
    ) \
    .select("user_id","session_start","session_end","event_count")

query = sessions.writeStream.outputMode("append").format("console").start()
query.awaitTermination()

Alternatively, Spark SQL:

sql

-- register events as view with watermark
CREATE OR REPLACE TEMP VIEW events AS
SELECT CAST(value AS STRING) JSON, from_json(CAST(value AS STRING),
  'event_id STRING, user_id STRING, event_time TIMESTAMP') AS d
FROM kafka_source;

-- use SQL with watermark and dedupe
SELECT user_id,
       session_start,
       session_end,
       COUNT(*) AS event_count
FROM (
  SELECT event_id, user_id, event_time
  FROM events
  -- apply watermark in SQL mode
) /* apply WITH WATERMARK and dropDuplicates via API */ 
GROUP BY user_id, SESSION(event_time, INTERVAL 30 MINUTES) AS (session_start, session_end)

Sample input (ordered for clarity):- {event_id: "e1", user_id: "u1", event_time: "2025-01-01T10:00:00Z"}- {event_id: "e2", user_id: "u1", event_time: "2025-01-01T10:05:00Z"}- {event_id: "e2", user_id: "u1", event_time: "2025-01-01T10:05:00Z"} -- duplicate- {event_id: "e3", user_id: "u1", event_time: "2025-01-01T11:00:00Z"} -- gap 55m -> new session- {event_id: "e4", user_id: "u2", event_time: "2025-01-01T10:10:00Z"}

Expected output:- user_id: u1, session_start: 2025-01-01T10:00:00, session_end: 2025-01-01T10:05:00, event_count: 2- user_id: u1, session_start: 2025-01-01T11:00:00, session_end: 2025-01-01T11:00:00, event_count: 1- user_id: u2, session_start: 2025-01-01T10:10:00, session_end: 2025-01-01T10:10:00, event_count: 1

Key points & caveats:- Watermark must be set before dropDuplicates to bound state and handle late arrivals.- dropDuplicates(["event_id"]) keeps first seen per event_id within watermark.- session_window (or SESSION in SQL) handles inactivity gaps; choose gap = '30 minutes'.- Choose watermark (e.g., 1 hour) based on expected lateness; too small may drop late-but-valid events, too large increases state.- Monitor state size; consider TTL or compaction for very high-cardinality user keys.

HardTechnical

40 practiced

Explain schema evolution strategies with Avro and a schema registry: backward compatibility, forward compatibility, and full compatibility. Provide a concrete example of a breaking change and how to modify the schema or consumers to restore compatibility.

Sample Answer

Backward, forward and full compatibility describe how reader and writer schemas can change over time without breaking consumers. With Avro + a schema registry (Confluent), the registry enforces a compatibility mode when new writer schemas are registered.

Definitions:- Backward compatible: New schema can read data written with the previous schema (new consumers can read old data). Registry mode = BACKWARD.- Forward compatible: Old consumers (old readers) can read data written by the new schema (new writers produce data old readers can read). Registry mode = FORWARD.- Full compatible: Both backward and forward (both old and new can read each other’s data). Registry mode = FULL.

Concrete example — breaking change:Writer v1 schema:

json

{
  "type":"record","name":"User","fields":[
    {"name":"id","type":"string"},
    {"name":"email","type":"string"}
  ]
}

If you remove "email" in v2:

json

{"type":"record","name":"User","fields":[{"name":"id","type":"string"}]}

This is NOT backward compatible: a new reader expecting only id cannot read old data that contains email? (Actually Avro ignores extra writer fields unless reader expects it.) Better breaking example: change type of "id" from string to int — that breaks both backward and forward.

Breaking change — type change:v1:

json

{"name":"id","type":"string"}

v2:

json

{"name":"id","type":"int"}

How to restore compatibility:- Prefer non-breaking alternatives: add new optional field instead of changing type:

json

{"name":"id_int","type":["null","int"],"default":null}

- Use a union to accept both types at read time (reader schema):Reader that tolerates both:

json

{"name":"id","type":["int","string"],"default":0}

- Use aliases when renaming fields:

json

{"name":"email_address","type":"string","aliases":["email"]}

- Provide defaults for removed fields so new readers can read old data:If you remove a field, set a default in the new schema for that field name (or keep it with default and mark deprecated).

Operational options:- Change consumers to handle the new shape (e.g., accept nulls, parse both id types).- Register schemas with appropriate compatibility mode in Schema Registry and run CI tests that validate consumers.- When forced type migration is needed, perform a controlled data migration job that rewrites historical data to the new schema.

Key reasoning:- Avro resolves differences using reader+writer schemas; defaults, unions, and aliases are the primary tools to ensure compatibility without rewriting data.- Choose compatibility policy based on consumer churn: BACKWARD is common so new consumers can roll out without needing producers to change. Full is strict but safest for bidirectional evolution.

MediumBehavioral

44 practiced

Tell me about a time you worked with source-system owners to reduce the operational impact of your ingestion pipeline (for example, to reduce locks or CPU load on a transactional DB). Describe the situation, the negotiation or technical changes you proposed, the outcome, and what you learned.

Unlock Full Question Bank

Get access to hundreds of Data Ingestion and Source Systems interview questions and detailed answers.

Join thousands of developers preparing for their dream job.