Assume a pipeline suddenly processes duplicate records for two days. How would you detect and fix this issue? What approach would you take?

First, I would add a unique identifier to each record (e.g., a hash of key fields) and store it in a deduplication table or use a distributed cache. To detect duplicates, I would run a query that groups by the unique key and counts occurrences, flagging any count > 1. For real‑time detection, I could use a windowed aggregation in Spark Structured Streaming or Flink that emits a warning when a key appears twice within a short window. Once duplicates are identified, I would trace back to the source system to understand why the same data was ingested twice—common causes include retry logic or misconfigured batch jobs. To fix the pipeline, I would make the ingestion idempotent: use upsert semantics or a merge operation that updates existing rows instead of inserting duplicates. Additionally, I would add a deduplication step in the ETL flow that filters out records whose unique key already exists in the target store. Finally, I would implement monitoring alerts on the duplicate detection metrics so that any future spikes are caught early.