Structured Streaming integration with Kafka | Data Engineering

Q: How do you read from Kafka?

You configure format("kafka") with the bootstrap servers and a subscription. Spark exposes each Kafka record as a row with key and value as binary, plus topic, partition, offset, and timestamp. You typically cast the value to a string and parse it from JSON or Avro. import org.apache.spark.sql.Datas

Q: How do you write to Kafka?

Prepare a DataFrame with a value column (and optionally key, topic, partition, headers) and call writeStream.format("kafka") with the bootstrap servers and a target topic. import java.math.BigDecimal; Dataset<Row enriched = transactions .filter(col("amount").gt(new BigDecimal("50000.00"))) .select(

Q: How do you control back-pressure?

The maxOffsetsPerTrigger option caps the total number of records read across all partitions per trigger. This is the primary backpressure control for a Kafka source, preventing a single batch from ingesting a massive backlog. // Without rate limit a 10M record backlog becomes one giant batch Dataset

Q: How is this different from Kafka Streams?

// Structured Streaming: clusterscale, SQLnative transactions.groupBy("accountId").count(); // Kafka Streams: lightweight, embedded // KStream<String, Transaction stream = builder.stream("nbitransactions"); // stream.groupByKey().count();

Q: What are common mistakes?

The most common mistake is claiming exactlyonce without making the sink idempotent. Spark guarantees it will replay uncommitted batches, but if the sink creates duplicates on replay, the endtoend result is atleastonce. // WRONG nonidempotent sink, duplicates on replay counts.writeStream() .foreach(n

Q: What should a senior engineer know?

Endtoend exactlyonce is a property of the whole chain, not just Spark. Spark guarantees it will reprocess uncommitted batches on recovery. The sink must guarantee that reprocessing does not produce duplicate effects. For a KafkatoKafka pipeline that means idempotent producers plus idempotent consume

Definition

Use `readStream.format("kafka")` to consume a topic as an unbounded DataFrame and `writeStream.format("kafka")` to produce back; checkpointed offsets give exactly-once on the source side, and idempotent or transactional sinks give exactly-once end-to-end.

Kafka source – spark.readStream.format("kafka") with bootstrap servers and a subscription reads a Kafka topic as an unbounded DataFrame with columns key, value, topic, partition, offset, and timestamp.
Kafka sink – df.writeStream.format("kafka") with bootstrap servers and a target topic writes rows back to Kafka; you prepare value and optionally key columns.
Offset management – Structured Streaming tracks consumed offsets in its own checkpoint, not in Kafka's consumer group, enabling deterministic replay.
Back-pressure – maxOffsetsPerTrigger caps records read per trigger, preventing a single batch from ingesting a massive backlog.
Exactly-once – Checkpointed offsets give exactly-once source semantics; end-to-end requires the sink to be idempotent (idempotent Kafka producers) or transactional.

At the National Bank of India Kafka is the central mail room where every transaction message is dropped and picked up. Structured Streaming's Kafka source is a clerk who reads mail from a named mailbox (topic), remembering in a logbook exactly which letters have been read so that after a break they resume at the right letter. The Kafka sink is a clerk who posts processed results back into another mailbox, stamping each letter with a unique id so the mail room ignores accidental resends.

Understanding the Concepts

How do you read from Kafka?

You configure format("kafka") with the bootstrap servers and a subscription. Spark exposes each Kafka record as a row with key and value as binary, plus topic, partition, offset, and timestamp. You typically cast the value to a string and parse it from JSON or Avro.

JAVA

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import static org.apache.spark.sql.functions.*;

SparkSession spark = SparkSession.builder()
    .appName("NBI-KafkaConsumer")
    .getOrCreate();

Dataset<Row> raw = spark.readStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
    .option("subscribe", "nbi-transactions")
    .option("startingOffsets", "earliest")
    .option("maxOffsetsPerTrigger", 10000)  // back-pressure
    .load();

// Kafka columns: key, value, topic, partition, offset, timestamp
Dataset<Row> transactions = raw
    .selectExpr("CAST(value AS STRING) as json")
    .select(from_json(col("json"), txnSchema).as("txn"))
    .select("txn.*");

TEXT

Kafka record columns:
+------+----------------+----------------+---------+------+-----------+
| key  | value (binary) | topic          |partition|offset| timestamp |
+------+----------------+----------------+---------+------+-----------+
| null | {"accountId"..} | nbi-transactions|    0   | 1042 | 1720000000|
+------+----------------+----------------+---------+------+-----------+

How does Spark manage Kafka offsets?

Structured Streaming tracks consumed offsets in its own checkpoint, not in Kafka's consumer group. This is what makes replay deterministic and exactly-once possible. On restart Spark resumes from its checkpointed offsets, so it neither skips nor reprocesses committed batches.

JAVA

// Offset management is automatic via checkpoint
transactions.writeStream()
    .outputMode("append")
    .format("console")
    .option("checkpointLocation", "s3a://nbi-checkpoints/kafka-consumer")
    .start();

// On restart:
// 1. Spark reads checkpoint: last committed offset = 5200
// 2. Resumes reading from offset 5201
// 3. Does NOT rely on Kafka consumer group offsets

How do you write to Kafka?

Prepare a DataFrame with a value column (and optionally key, topic, partition, headers) and call writeStream.format("kafka") with the bootstrap servers and a target topic.

JAVA

import java.math.BigDecimal;

Dataset<Row> enriched = transactions
    .filter(col("amount").gt(new BigDecimal("50000.00")))
    .select(
        col("accountId").as("key"),
        to_json(struct("accountId", "amount", "eventTime")).as("value")
    );

enriched.writeStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("topic", "nbi-high-value-alerts")
    .option("checkpointLocation", "s3a://nbi-checkpoints/kafka-sink")
    .start();

How do you control back-pressure?

The maxOffsetsPerTrigger option caps the total number of records read across all partitions per trigger. This is the primary back-pressure control for a Kafka source, preventing a single batch from ingesting a massive backlog.

JAVA

// Without rate limit -- a 10M record backlog becomes one giant batch
Dataset<Row> raw = spark.readStream()
    .format("kafka")
    .option("subscribe", "nbi-transactions")
    .load(); // May read millions in one batch!

// With rate limit -- backlog is drained in bounded chunks
Dataset<Row> bounded = spark.readStream()
    .format("kafka")
    .option("subscribe", "nbi-transactions")
    .option("maxOffsetsPerTrigger", 50000) // max 50K records per batch
    .load();

What delivery semantics does it provide?

The Kafka source with checkpointed offsets gives effectively exactly-once consumption. For end-to-end exactly-once, the sink must not create visible duplicates on retry. With the Kafka sink you enable idempotent producers. With a database sink you use an upsert by primary key.

JAVA

// Kafka sink with idempotent producer
enriched.writeStream()
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092")
    .option("topic", "nbi-enriched-txns")
    .option("kafka.enable.idempotence", "true")  // idempotent producer
    .option("checkpointLocation", "s3a://nbi-checkpoints/enriched")
    .start();

// Database sink with foreachBatch upsert
counts.writeStream()
    .foreachBatch((batchDf, batchId) -> {
        batchDf.write()
            .format("jdbc")
            .option("url", "jdbc:postgresql://nbi-db:5432/analytics")
            .option("dbtable", "account_counts")
            .mode("append")  // with ON CONFLICT DO UPDATE
            .save();
    })
    .option("checkpointLocation", "s3a://nbi-checkpoints/db-upsert")
    .start();

How is this different from Kafka Streams?

| Aspect | Structured Streaming + Kafka | Kafka Streams |
|---|---|---|
| Engine | Spark (distributed batch engine) | Kafka client library |
| Deployment | Cluster (YARN/K8s/standalone) | Embedded in JVM app |
| Latency | Micro-batch (100ms-seconds) | Record-by-record (ms) |
| State | Spark state store / RocksDB | RocksDB per partition |
| SQL support | Full Spark SQL | KSQL (separate) |
| Best for | Complex analytics, ML enrichment | Simple event-driven transforms |

JAVA

// Structured Streaming: cluster-scale, SQL-native
transactions.groupBy("accountId").count();

// Kafka Streams: lightweight, embedded
// KStream<String, Transaction> stream = builder.stream("nbi-transactions");
// stream.groupByKey().count();

What are common mistakes?

The most common mistake is claiming exactly-once without making the sink idempotent. Spark guarantees it will replay uncommitted batches, but if the sink creates duplicates on replay, the end-to-end result is at-least-once.

JAVA

// WRONG -- non-idempotent sink, duplicates on replay
counts.writeStream()
    .foreach(new ForeachWriter<Row>() {
        public boolean open(long partitionId, long epochId) { return true; }
        public void process(Row row) {
            // INSERT into database without ON CONFLICT
            // Replay creates duplicate rows!
        }
        public void close(Throwable error) {}
    })
    .start();

// CORRECT -- idempotent upsert
counts.writeStream()
    .foreachBatch((df, batchId) -> {
        // UPSERT with ON CONFLICT DO UPDATE
        df.write().format("jdbc")
            .option("dbtable", "account_counts")
            .mode("append").save();
    }).start();

What should a senior engineer know?

End-to-end exactly-once is a property of the whole chain, not just Spark. Spark guarantees it will reprocess uncommitted batches on recovery. The sink must guarantee that reprocessing does not produce duplicate effects. For a Kafka-to-Kafka pipeline that means idempotent producers plus idempotent consumers keyed on a stable business id such as a transaction id. Designing the sink for idempotency is the real work of exactly-once.

Visual Overview

DIAGRAM

  Structured Streaming + Kafka Pipeline

  ┌─────────────────────────────────────────────────┐
  │  Kafka Cluster                                   │
  │  ┌──────────────────┐  ┌──────────────────┐     │
  │  │ nbi-transactions │  │ nbi-enriched-txns│     │
  │  │ (source topic)   │  │ (sink topic)     │     │
  │  │ P0 P1 P2 P3     │  │ P0 P1 P2 P3     │     │
  │  └────────┬─────────┘  └────────▲─────────┘     │
  └───────────┼─────────────────────┼───────────────┘
              │                     │
              ▼                     │
  ┌───────────────────────┐         │
  │ readStream             │         │
  │ .format("kafka")       │         │
  │ .option(subscribe)     │         │
  │ .option(maxOffsets)    │         │
  └───────────┬───────────┘         │
              │                     │
              ▼                     │
  ┌───────────────────────┐         │
  │ DataFrame Transform    │         │
  │ filter / groupBy /     │         │
  │ withWatermark / join   │         │
  └───────────┬───────────┘         │
              │                     │
              ▼                     │
  ┌───────────────────────┐         │
  │ writeStream            │─────────┘
  │ .format("kafka")       │
  │ .option(idempotent)    │
  └───────────┬───────────┘
              │
              ▼
  ┌───────────────────────┐
  │ Checkpoint (S3)        │
  │ Offsets tracked here,  │
  │ NOT in Kafka consumer  │
  │ group                  │
  └───────────────────────┘

Key Takeaway

Structured Streaming integrates with Kafka through a built-in source and sink. As a source you use readStream.format("kafka") with the bootstrap servers and a subscription, and Spark exposes the topic as an unbounded DataFrame whose rows carry key, value, topic, partition, offset, and timestamp. You typically cast or parse the value from JSON or Avro and control intake with maxOffsetsPerTrigger for back-pressure.

The crucial point about offsets is that Structured Streaming tracks consumed offsets in its own checkpoint, not in Kafka's consumer group. That is what lets it resume exactly where it left off and makes replay deterministic, giving effectively exactly-once consumption of the source. But end-to-end exactly-once is a property of the whole chain: Spark guarantees it will reprocess a batch that was started but not committed, so the sink must guarantee that reprocessing does not create duplicate effects.

For a Kafka sink that means idempotent or transactional producers, and downstream consumers should still deduplicate on a stable business key such as the transaction id. For a database sink that means an upsert by primary key inside a foreachBatch. Designing the sink for idempotency is the real work of exactly-once.

At the National Bank of India Rajesh built a Kafka-to-Kafka enrichment job where the source consumed the transactions topic with offsets tracked in an S3 checkpoint, and the sink wrote to an enriched topic keyed by transaction id with idempotent producers. When a batch was retried after a transient failure, the duplicate was ignored downstream, so the effect was exactly-once.

Remember one sentence: readStream and writeStream with format kafka, offsets in Spark's checkpoint for exactly-once source semantics, and idempotent or transactional sinks for exactly-once end-to-end.

---