Ensuring Exact‑Once Semantics in Spark Streaming with Kafka: Offline Repair and Data Deduplication Strategies

Spark Streaming plays a pivotal role in many stream‑processing ecosystems, but network jitter often causes data delays that lead to duplicate consumption when jobs are updated or unexpectedly crash, making exact‑once guarantees difficult.

Two notions of time are essential: Event time (the timestamp attached to each event) and Processing time (the server clock when the event is processed). Because Kafka partitions do not preserve order, the same event time can appear in multiple batches, creating two main problems: a batch may contain data from several time windows, and the same time‑window data may be split across different batches.

For the first problem, the solution is to truncate each event’s timestamp to the nearest 5‑minute window and group records by this truncated time, then aggregate with a simple SELECT time, COUNT(*) GROUP BY truncated_time query.

The second problem—duplicate data across batches—cannot be solved reliably with UpdateStateByKey or in‑memory checkpoints because they introduce state management overhead and still cannot guarantee exact‑once. Instead, the article recommends persisting state externally (e.g., Redis or HBase) and using a unique identifier for each Kafka message, such as partition+offset , to enable deduplication.

If the sink is HBase, the unique ID becomes the row key, ensuring idempotent writes; if the sink is HDFS, prepend the ID to each line and deduplicate during offline processing.

When a job fails after a checkpoint, replaying the affected time window with the full offline data (stored by event‑time partitions) restores correctness. This approach requires maintaining a complete, non‑overlapping offline dataset that can be re‑processed for any problematic window.

Output to HDFS or HBase is performed via saveAsHadoopDataset . The following Scala snippet shows how Spark writes records using a SparkHadoopWriter :

val writer = new SparkHadoopWriter(hadoopConf)
writer.open()
Utils.tryWithSafeFinallyAndFailureCallbacks {
    while (iter.hasNext) {
        val record = iter.next()
        writer.write(record._1.asInstanceOf[AnyRef], record._2.asInstanceOf[AnyRef])
    }
}(finallyBlock = writer.close())
writer.commit()

For HBase, the writer ultimately calls the HBase client’s put method, which can be buffered and batch‑committed for performance. For HDFS, the writer creates a temporary path based on TaskAttemptID , writes data, and on commit renames the temporary file to the final destination, avoiding conflicts.

To output multiple files based on different keys, the article introduces MultipleOutputFormat . A TreeMap stores a RecordWriter for each key, and each writer uses getBaseRecordWriter → TextOutputFormat → LineRecordWriter → DataOutputStream to write data. By modifying the writer to perform an append instead of creating a new temporary file, small‑file proliferation on HDFS can be mitigated.

val fileOut: FSDataOutputStream = if (HDFSFileService.existsPath(file)) {
    println("appendfile")
    fs.append(file)
} else {
    println("createfile")
    fs.create(file, progress)
}

def getTaskOutputPath(job: JobConf, iname: String): Path = {
    val name: String = job.get(org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.OUTDIR)
    val completePath = name + "/" + iname
    new Path(completePath)
}

In summary, without modifying Spark’s core code, the proposed extensions uniquely identify Kafka messages, deduplicate them in persistent storage, and enable offline replay of erroneous time windows, thereby achieving exact‑once semantics for Spark Streaming applications.