Unit 5: Spark Streaming with Apache Kafka

I. Orientation

Apache Kafka is a distributed event-streaming platform designed to collect, store, and deliver continuously generated records. Apache Spark Streaming and Structured Streaming process these records as micro-batches or continuous streams. Together, Kafka provides durable, scalable transport while Spark performs computation, transformation, aggregation, and output.

• Governing principle: Producers write records to Kafka topics; consumers read those records independently and process them according to their offsets.
• Distributed design: Kafka partitions topics across brokers, while Spark distributes stream-processing tasks across executors.
• Durability assumption: A record remains available according to the topic retention policy, even after one consumer has read it.
• Processing convention: A Kafka record generally contains a key, a value, a timestamp, headers, a topic, a partition, and an offset.
• Fault-tolerance principle: Kafka replication and Spark checkpointing allow processing to resume after broker, executor, or application failures.
• Scalability condition: Parallelism depends on the number of Kafka partitions and the number of Spark tasks that can process them concurrently.

II. Fundamentals of Apache Kafka

A. Definition and purpose

Apache Kafka is a distributed publish-subscribe system that treats streams of records as durable, ordered logs. It is used for messaging, event sourcing, log collection, and real-time data integration.

• Record structure: A message may contain key = customer-42, value = {"amount":125}, and a timestamp.
• Topic role: A topic is a named logical stream such as transactions or sensor-readings.
• Partition ordering: Kafka guarantees ordering within one partition, but not automatically across all partitions of a topic.
• Retention: Records are retained by time or size, for example retention.ms or retention.bytes, rather than being deleted immediately after consumption.
• Pull-based consumption: Consumers request records from Kafka, allowing them to control read rate and handle temporary processing delays.
• Use cases: Kafka commonly connects web applications, databases, Spark jobs, monitoring systems, and data warehouses.

B. Fundamentals of Apache Kafka

The fundamentals determine how Kafka achieves high throughput and independent consumer progress.

• Append-only log: New records are appended to the end of a partition; each record receives a sequential offset such as 0, 1, or 2.
• Key-based routing: Records with the same key are normally sent to the same partition, preserving per-key order.
• Consumer group: Consumers with the same group.id divide partitions among themselves; one partition is assigned to only one active consumer in that group.
• Independent groups: A monitoring group and an archival group can both read the same topic without interfering with each other.
• Delivery behavior: Standard processing is commonly described as at-least-once, because a record may be processed again after a failure before its progress is safely committed.
• Throughput control: Batch size, compression, partition count, and broker resources affect the number of records processed per second.

III. Apache Kafka Cluster Architecture

A. Structure and operation

An Apache Kafka cluster is a group of broker servers cooperating to store and serve partitioned topic data. Modern Kafka deployments can use KRaft metadata management; older deployments used ZooKeeper.

• Broker: A broker stores topic partitions and handles produce and fetch requests. A cluster might contain brokers 1, 2, and 3.
• Partition leader: Each partition has one leader that handles client reads and writes.
• Replica: Followers replicate the leader’s log. With replication factor 3, three brokers store copies of each partition.
• In-sync replicas: The ISR contains replicas considered sufficiently caught up with the leader; acks=all requires acknowledgement from the relevant in-sync replicas.
• Controller: The controller manages partition leadership and broker membership. In KRaft mode, controller quorum metadata is managed without ZooKeeper.
• Rebalancing: When a broker fails or a consumer joins a group, partition assignments may be redistributed.
• Scalability limit: A topic with six partitions can provide up to six-way partition-level consumer parallelism within one consumer group.

B. Apache Kafka cluster architecture

Architecture choices affect availability, ordering, and recovery.

• Replication trade-off: Replication factor 3 tolerates broker loss better than factor 1, but consumes approximately three times the partition storage.
• Leader failure: A new leader is selected from eligible replicas, allowing clients to continue after recovery.
• Acknowledgement modes:
  • acks=0: the producer does not wait for broker confirmation.
  • acks=1: the leader acknowledges the write.
  • acks=all: the leader waits for the required replicas.
• Partition placement: Partitions should be distributed across brokers so that no single broker receives all traffic or storage.
• Ordering boundary: Increasing partitions improves parallelism but can make global ordering impossible.
• Retention versus deletion: A consumer’s position does not delete data; broker retention determines when old records disappear.

IV. Apache Kafka Installation

A. Purpose and principle

Kafka installation creates brokers, configures metadata management, and exposes listeners through which producers and consumers connect. A local development installation differs from a production cluster in security, replication, and operational requirements.

• Prerequisites: A compatible Java runtime is required for Kafka distributions; the exact supported version depends on the Kafka release.
• KRaft setup: Current Kafka releases can run in KRaft mode, where brokers and controllers use a cluster ID and formatted storage directories.
• Listener configuration: listeners defines bind addresses, while advertised.listeners defines addresses returned to clients.
• Port convention: Port 9092 is commonly used for a client listener, but production deployments should use explicit network and security settings.
• Storage path: log.dirs identifies where partition logs, indexes, and metadata are stored.
• Production requirements: Use multiple brokers, persistent disks, authentication, authorization, monitoring, and carefully planned replication.

B. Apache Kafka installation

A minimal KRaft-based installation can be represented by the following command sequence; options vary by Kafka release.

• Generate identity: A cluster UUID identifies the Kafka cluster and is used when formatting storage.
• Format storage: The storage tool initializes the log directory before the broker starts.
• Start broker: The server reads a properties file containing roles, listeners, log directories, and quorum settings.
• Create topic: A topic can be created with a defined partition and replication count.

KAFKA_CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)
bin/kafka-storage.sh format \
  --standalone \
  -t "$KAFKA_CLUSTER_ID" \
  -c config/kraft/reconfig-server.properties

bin/kafka-server-start.sh config/kraft/reconfig-server.properties

bin/kafka-topics.sh --create \
  --topic events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

• Verification: kafka-topics.sh --describe --topic events --bootstrap-server localhost:9092 displays partitions, leaders, and replicas.
• Operational caution: A single-node broker with replication factor 1 is suitable for learning, not for fault-tolerant production service.

V. Integration of Apache Kafka with Spark

A. Purpose and principle

Integration of Apache Kafka with Spark allows Spark applications to consume Kafka records, deserialize their values, transform the resulting data, and write results to another system or Kafka topic.

• Connector dependency: Spark requires the matching spark-sql-kafka-0-10 connector for the Spark and Scala versions in use.
• Input columns: A Kafka source exposes fields including key, value, topic, partition, offset, timestamp, and timestampType.
• Binary data: Kafka key and value arrive as binary values and normally require casting or deserialization.
• Structured Streaming source: Spark treats Kafka as an unbounded input table and processes newly available records incrementally.
• Offsets: Spark manages source progress through checkpoint data rather than relying only on Kafka consumer commits.
• Output modes: append, update, or complete determine how results are written, depending on the query and sink.

B. Integration of Apache Kafka with Spark

A typical Spark application reads a Kafka topic, converts JSON values to typed columns, and writes a result.

• Read configuration: subscribe selects topics; startingOffsets may be earliest or latest.
• Schema control: An explicit schema prevents ambiguous types such as treating an amount as a string.
• Checkpointing: checkpointLocation stores progress and state needed for recovery.

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("customer", StringType()) \
    .add("amount", DoubleType())

events = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "transactions")
    .option("startingOffsets", "latest")
    .load())

typed = (events
    .select(F.from_json(F.col("value").cast("string"), schema).alias("data"))
    .select("data.*"))

query = (typed.writeStream
    .format("console")
    .outputMode("append")
    .option("checkpointLocation", "/tmp/transactions-checkpoint")
    .start())

• Parallelism mapping: Kafka partitions provide input parallelism, although Spark may coalesce or repartition data during transformations.
• Backpressure: maxOffsetsPerTrigger limits records read in one micro-batch and helps control processing load.
• Security: Kafka properties such as SASL and SSL settings must be supplied to Spark when the cluster is secured.

VI. Producer and Consumer Messaging Model

A. Definition and operation

The producer and consumer messaging model separates message publication from message processing. Producers send records to topics, while consumers fetch records and track their positions.

• Producer responsibility: A producer serializes keys and values and selects a topic partition.
• Consumer responsibility: A consumer deserializes records and processes them according to application logic.
• Offset meaning: Offset 15 identifies a position within one partition; it is not a globally unique message ID.
• Consumer group assignment: If topic orders has four partitions and group G has two consumers, each consumer may receive two partitions.
• Commit behavior: Committing an offset records how far a consumer group has progressed.
• Duplicate handling: A crash after processing but before committing can cause the same record to be processed again.

B. Producer and consumer messaging model

Reliability depends on how acknowledgements, retries, commits, and processing are coordinated.

• Producer reliability: enable.idempotence=true helps prevent duplicate writes caused by producer retries.
• Batching: batch.size and linger.ms allow records to be sent in batches, improving throughput at the cost of small latency.
• Consumer polling: Consumers repeatedly call poll() to obtain records and must continue polling within the configured interval.
• At-least-once processing: Commit offsets after successful processing so a failure causes replay rather than silent loss.
• Exactly-once boundaries: Kafka transactions and Spark checkpointing can provide stronger guarantees, but the complete source-to-sink path must support them.
• Poison records: Invalid JSON or schema violations should be routed to a dead-letter topic or error store instead of repeatedly blocking progress.

VII. Kafka Pipeline

A. End-to-end flow

A Kafka pipeline is the ordered movement of event data from its origin through Kafka and processing systems to a destination. Each stage has a distinct responsibility.

• Event source: A web service, application log, database change stream, or IoT device generates an event.
• Producer stage: The producer validates, serializes, and publishes the event to a topic.
• Kafka stage: Brokers partition, replicate, retain, and serve the record.
• Processing stage: Spark reads records, parses fields, filters invalid data, joins reference data, and computes aggregates.
• Sink stage: Results may be written to Kafka, a database, object storage, or a dashboard system.
• Monitoring stage: Metrics such as consumer lag, throughput, failed records, and processing latency reveal pipeline health.

B. Kafka pipeline

A robust pipeline preserves data contracts and makes failures observable.

• Schema contract: An event such as {"id":"A7","amount":125.0} should have documented field names, types, and compatibility rules.
• Transformation boundary: Raw events can be stored in transactions.raw, while validated records move to transactions.cleaned.
• Lag measurement: Consumer lag is the difference between the latest partition offset and the consumer group’s committed offset.
• Failure handling: Retryable failures may be retried; malformed records should be isolated with the original payload and error reason.
• Idempotent sink: Writing with a stable event ID, such as id = A7, allows a sink to ignore repeated deliveries.
• Data lifecycle: Retention, compaction, encryption, and access permissions should match the business and compliance requirements.

VIII. Ingestion Patterns

A. Purpose and principle

Ingestion patterns describe how data enters Kafka and how Spark consumes it. The appropriate pattern depends on latency, replay needs, source behavior, and delivery guarantees.

• Direct event ingestion: Applications publish events immediately after an action, such as a completed payment.
• Log or file ingestion: Agents collect records from files and forward them to Kafka, often preserving source offsets.
• Change data capture: A connector reads database changes and publishes insert, update, and delete events.
• Batch-to-stream ingestion: Periodic jobs publish accumulated records, producing lower operational complexity but higher latency.
• Fan-in: Many producers send to one topic, such as events, where a common processing layer handles them.
• Fan-out: One topic feeds several independent consumer groups, such as analytics, fraud detection, and archival services.

B. Ingestion patterns

Choosing an ingestion pattern requires balancing freshness, ordering, throughput, and recovery.

• Event-time ingestion: Use the timestamp attached to the event, such as 2025-04-10 12:00:05, when measuring when an action actually occurred.
• Processing-time ingestion: Use the time Spark receives the record when arrival latency is more important than source-time accuracy.
• Late data: Spark watermarks bound how long the engine waits for delayed events; a one-hour watermark may exclude events arriving more than one hour late from a finalized window.
• Partitioning strategy: Partition by a stable key such as account_id to preserve account order, while avoiding a single extremely active key.
• Replay pattern: Resetting or selecting a consumer offset allows historical records to be reprocessed, provided Kafka retention still contains them.
• Exactly-once-oriented pattern: Combine deterministic transformations, checkpoints, transactional Kafka writes, and idempotent external sinks; no single setting guarantees end-to-end exactly-once behavior.