Unit 5: Spark Streaming with Apache Kafka - Subjective Questions

Apache Kafka is a distributed event-streaming platform used to publish, store, and process continuous streams of records in real time.

Key characteristics:

• Distributed: Kafka runs across multiple cluster nodes called brokers.
• Scalable: Topics can be divided into partitions and distributed among brokers.
• Fault-tolerant: Partition replication protects data against broker failures.
• High throughput: Kafka can process large volumes of messages with low latency.
• Durable: Messages are written to disk and retained according to a configured retention policy.

Importance in cluster computing:

• It provides a reliable communication layer between distributed applications.
• It decouples data producers from data consumers.
• It supports parallel processing through topic partitions.
• It integrates with processing engines such as Apache Spark for real-time analytics.

Thus, Kafka acts as a scalable data backbone for log collection, monitoring, event processing, and streaming analytics.

The fundamental concepts of Apache Kafka are:

• Message or record: A unit of data stored in Kafka. It commonly contains a key, value, timestamp, and optional headers.
• Topic: A named logical stream to which messages are published. For example, transactions may store transaction events.
• Partition: A topic is divided into one or more partitions. Each partition is an ordered, append-only sequence of records.
• Offset: A unique sequential number assigned to each record within a partition. Consumers use offsets to track their reading position.
• Broker: A Kafka server responsible for storing partitions and serving producer and consumer requests.
• Producer: An application that publishes records to Kafka topics.
• Consumer: An application that reads and processes records from topics.

Kafka guarantees record ordering within an individual partition, but it does not provide global ordering across all partitions of a topic.

An Apache Kafka cluster consists of multiple cooperating brokers that store and serve streaming data.

Architectural components:

1. Brokers: Each broker has a unique identifier and stores a subset of the topic partitions.
2. Topics: Messages are organized into named topics.
3. Partitions: Each topic is split into partitions, allowing storage and processing to be distributed across brokers.
4. Partition leader: One broker acts as the leader for each partition. All reads and writes for that partition are handled by its leader.
5. Partition followers: Other brokers hold replicas and copy records from the leader.
6. Controller: A broker acting as the controller coordinates partition leadership and reacts to broker failures.
7. Metadata management: Modern Kafka uses the KRaft consensus mechanism and controller quorum for metadata management. Older Kafka deployments may use ZooKeeper.

Operation:

• Producers discover the partition leader and send records to it.
• Followers replicate the leader's log.
• Consumers fetch records from the relevant partitions.
• If a leader fails, an eligible in-sync follower can be elected as the new leader.

This architecture provides scalability, parallelism, data durability, and fault tolerance.

Partitioning divides a Kafka topic into multiple ordered logs that can be placed on different brokers.

Effect on scalability:

• Partitions distribute data and request load across the cluster.
• Increasing partitions can allow a topic to use more brokers and handle greater throughput.

Effect on ordering:

• Kafka maintains strict ordering only within a partition.
• Records with the same key are normally directed to the same partition, preserving their relative order.
• No total ordering is guaranteed across different partitions.

Effect on parallelism:

• Different partitions can be consumed simultaneously.
• Within one consumer group, a partition is assigned to at most one consumer at a time.
• Therefore, the effective parallelism of a consumer group is limited by the number of partitions.

If a topic has partitions and a consumer group has consumers, the maximum number of actively assigned consumers is:

An appropriate partition count must balance throughput, ordering requirements, resource usage, and operational overhead.

Kafka achieves fault tolerance by maintaining multiple replicas of each topic partition on different brokers.

• Replication factor: Specifies the number of copies of a partition. A replication factor of 3 creates one leader replica and two follower replicas.
• Leader replica: Handles producer writes and normally serves consumer reads for the partition.
• Follower replicas: Continuously fetch records from the leader and maintain copies of its log.
• In-sync replicas (ISR): Replicas that are sufficiently caught up with the leader are included in the ISR set.

Failure handling:

• If the leader broker fails, the controller can elect an eligible in-sync follower as the new leader.
• Replicas should be distributed across brokers so that one broker failure does not remove all copies.
• Producer configuration such as acks=all can require acknowledgement from the required in-sync replicas.
• min.insync.replicas specifies the minimum ISR count needed for successful durable writes when strong acknowledgement is requested.

A larger replication factor improves availability and durability but increases storage consumption and network replication traffic.

The major installation and configuration steps are:

1. Meet prerequisites:
   • Install a supported Java Development Kit.
   • Provide sufficient disk space, memory, and network connectivity.
2. Obtain Kafka:
   • Download and extract an Apache Kafka binary distribution.
3. Configure the servers:
   • Assign a unique broker or node identifier.
   • Configure listener addresses and advertised listeners.
   • Set log directories, default partition count, and retention settings.
4. Configure metadata mode:
   • For modern deployments, configure KRaft controller and broker roles.
   • Create a cluster identifier and format the storage directories.
   • Older versions may require a separately managed ZooKeeper service.
5. Start the cluster:
   • Start controller nodes where applicable.
   • Start Kafka brokers.
6. Create and verify a topic:
   • Create a topic with suitable partitions and replication factor.
   • List or describe the topic to confirm leader and replica assignments.
7. Test communication:
   • Use a console producer to send records.
   • Use a console consumer to read records.
8. Prepare production controls:
   • Enable authentication, authorization, encryption, monitoring, and log management.

A multi-broker installation should place replicas on separate machines for genuine fault tolerance.

Important Kafka configuration parameters include:

• Node or broker identifier: Uniquely identifies a server in the cluster.
• listeners: Defines the network interfaces and ports on which Kafka accepts connections.
• advertised.listeners: Specifies the addresses provided to clients. These must be reachable from producer and consumer machines.
• log.dirs: Identifies directories in which partition logs are stored.
• num.partitions: Sets the default number of partitions for newly created topics.
• default.replication.factor: Determines the default number of replicas where supported by topic creation settings.
• log.retention.hours or related settings: Determines how long messages are retained.
• log.segment.bytes: Controls the size of log segment files.
• min.insync.replicas: Sets the minimum number of in-sync replicas required for strongly acknowledged writes.
• KRaft settings: Parameters such as process roles, node ID, controller listener, and controller quorum voters configure metadata management.

These settings influence accessibility, storage usage, performance, durability, and availability. Incorrect advertised listeners are a common reason clients cannot connect to a newly installed cluster.

A Kafka producer publishes records to one or more topics through the following process:

1. Record creation: The application creates a record containing a topic, optional partition, optional key, value, timestamp, and headers.
2. Serialization: Key and value serializers convert application objects into byte arrays.
3. Partition selection:
   • An explicitly specified partition is used directly.
   • A keyed record is generally mapped consistently using the key.
   • Records without keys may be distributed among available partitions.
4. Batching: Records for the same partition are grouped into batches to reduce network requests and improve throughput.
5. Compression: Batches may be compressed using a configured codec.
6. Transmission: The producer sends the batch to the partition leader.
7. Acknowledgement:
   • acks=0 does not wait for confirmation.
   • acks=1 waits for the leader.
   • acks=all waits for acknowledgement according to the in-sync replica requirements.
8. Retries: Retriable failures may cause automatic resubmission.

Idempotent production can prevent duplicate writes caused by retries and helps provide reliable delivery.

A Kafka consumer subscribes to topics, receives partition assignments, and fetches records from the corresponding brokers.

Consumer operation:

• The consumer subscribes to one or more topics.
• It joins a consumer group when a group identifier is configured.
• Partitions are assigned among the consumers in the group.
• The consumer repeatedly polls Kafka for records.
• After processing, it records its progress using offsets.

Offset management:

• An offset represents the position of a record within a partition.
• A committed offset normally indicates where the consumer group should resume processing.
• Automatic commit periodically commits offsets but may not match actual processing completion.
• Manual commit allows the application to commit only after successful processing.

If an offset is committed before processing finishes, a failure can cause data loss from the application's perspective. If it is committed after processing, a failure before the commit can lead to repeated processing. Reliable applications therefore coordinate processing, output, and offset commitment carefully.

A consumer group is a collection of consumers that cooperate to process records from one or more topics.

Partition assignment:

• Kafka assigns each subscribed partition to one consumer within the group.
• A partition is not processed concurrently by multiple consumers in the same group.
• Different consumer groups can independently consume the same topic.

Parallel consumption:

• If a topic has multiple partitions, consumers in the group can process them in parallel.
• When consumers exceed the number of partitions, some consumers remain idle.
• One consumer may receive multiple partitions when there are fewer consumers than partitions.

Rebalancing:

• A rebalance may occur when a consumer joins or leaves, a consumer fails, or the topic's partition count changes.
• During rebalancing, partitions are revoked and reassigned.
• Poorly controlled rebalances can temporarily pause processing and may increase duplicate work.

Applications should poll regularly, process records within configured time limits, and use appropriate assignment strategies to minimize disruption.

Discussion:

• At-most-once favors low duplication risk but permits loss.
• At-least-once is common because it prevents silent loss, but output operations should be idempotent or support deduplication.
• Exactly-once is not achieved merely by setting one Kafka consumer parameter. It requires end-to-end support from the source, processing engine, and sink.
• Kafka transactions and idempotent producers can support exactly-once behavior for Kafka-to-Kafka processing.
• In Spark Structured Streaming, checkpointing and sink-specific transactional or idempotent mechanisms are needed for reliable end-to-end results.

Spark Structured Streaming integrates with Kafka by using Kafka as a streaming source, a sink, or both.

Kafka as a source:

1. Add the Spark-Kafka connector package compatible with the Spark and Scala versions.
2. Create a streaming DataFrame with readStream.
3. Set the format to kafka.
4. Configure broker addresses through kafka.bootstrap.servers.
5. Select topics using subscribe, assign, or subscribePattern.
6. Convert Kafka's binary key and value columns into the required data types.
7. Parse, filter, aggregate, or enrich the data.
8. Write the result using writeStream and configure a checkpoint location.

Kafka as a sink:

• The output must provide a value column and may provide key and topic columns.
• Spark serializes these columns as binary data and publishes them to Kafka.

Kafka provides scalable message transport, while Spark performs distributed transformations, stateful processing, windowing, and analytics.

A typical PySpark workflow is:

source = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "broker1:9092").option("subscribe", "input-topic").load()

events = source.selectExpr("CAST(key AS STRING) AS key", "CAST(value AS STRING) AS value")

processed = events.selectExpr("CAST(key AS STRING) AS key", "CAST(upper(value) AS STRING) AS value")

query = processed.writeStream.format("kafka").option("kafka.bootstrap.servers", "broker1:9092").option("topic", "output-topic").option("checkpointLocation", "/checkpoints/kafka-query").start()

Explanation:

• readStream creates a streaming Kafka source.
• kafka.bootstrap.servers identifies the initial brokers used for metadata discovery.
• subscribe specifies the input topic.
• Kafka keys and values are initially represented as binary columns, so they are cast before processing.
• The transformation converts the message value to uppercase as a simple example.
• writeStream configures Kafka as the destination.
• The output includes columns named key and value, which Kafka expects.
• checkpointLocation stores progress and state information so that the query can recover after failure.
• start() begins execution, while the application commonly uses awaitTermination() to remain active.

Structured Streaming provides a unified batch and streaming API and is generally easier to integrate with parsing, SQL analytics, aggregations, and multiple sinks. DStreams remain relevant when maintaining older Spark Streaming applications.

A complete Kafka-Spark pipeline can contain the following stages:

1. Data sources: Sensors, applications, web servers, databases, or devices generate events.
2. Kafka producers: Producers serialize the events and publish them to input topics.
3. Kafka ingestion layer: Brokers store records in partitioned and replicated logs, buffering the difference between production and processing rates.
4. Spark source: Spark Structured Streaming reads topic partitions in parallel.
5. Parsing and validation: Binary values are deserialized and checked against the expected schema.
6. Transformation: Spark filters, enriches, joins, aggregates, or applies event-time windows.
7. State and recovery: Checkpoints, offset logs, and state stores support recovery and stateful operations.
8. Output sinks: Results may be sent to another Kafka topic, a data lake, a database, a search system, or an alerting service.
9. Serving and visualization: Dashboards and analytical tools present trends, alerts, and operational metrics.
10. Monitoring: Kafka lag, broker health, Spark batch duration, input rate, processing rate, and failures are tracked.

The pipeline is scalable because Kafka partitions ingestion and Spark distributes computation across cluster executors.

Ingestion patterns describe how data from external systems is captured and introduced into Kafka.

Common patterns include:

• Direct producer ingestion: An application uses the Kafka producer API to publish events immediately.
• Log and file ingestion: Agents or connectors monitor files and forward new log entries to Kafka.
• Database change data capture (CDC): Insert, update, and delete operations are captured from database logs and published as events.
• Polling ingestion: A connector periodically queries an API or database for new records. This is simpler than CDC but may increase latency and load.
• Batch-to-stream ingestion: Existing files or historical records are read in batches and published to Kafka.
• IoT and telemetry ingestion: Gateways collect device readings and send them to partitioned topics.
• Fan-in pattern: Many producers send records to a shared topic or topic family.
• Multi-stage ingestion: Raw events enter one topic and are validated or normalized before being written to a clean topic.

Pattern selection depends on latency, data volume, ordering, reliability, source capabilities, and schema requirements.

Push-based ingestion:

• The source application actively publishes records to Kafka.
• It usually uses the Kafka producer API or an event gateway.
• It provides low latency because records are sent when events occur.
• It requires producer logic, retry handling, serialization, and security configuration in or near the source.

Pull-based ingestion:

• A connector or ingestion service periodically retrieves data from the source.
• It is useful when the source cannot publish directly to Kafka.
• It can work with databases, REST APIs, files, and legacy systems.
• Its latency depends on the polling interval.
• It must track the last imported position to avoid missed or duplicate records.

Comparison:

• Push is suitable for event-driven applications and immediate delivery.
• Pull is suitable for passive or legacy data sources.
• Both approaches require handling failures, backpressure, schemas, duplicates, and source progress.
• Kafka consumers themselves use a pull-oriented model to fetch records, which allows them to control their consumption rate.

Consumer lag is the difference between the latest offset in a Kafka partition and the offset processed or committed by a consumer. For partition :

Total group lag can be expressed as:

Backpressure occurs when data arrives faster than Spark or another consumer can process it. Sustained backpressure causes lag to grow.

Throughput is the amount of data processed per unit time, such as records per second or bytes per second.

Methods for addressing bottlenecks:

• Increase topic partitions to permit more parallel consumers.
• Add Spark executors, cores, or memory when computation is the bottleneck.
• Optimize transformations, joins, serialization, and stateful operations.
• Use producer batching and compression appropriately.
• Tune Spark trigger intervals and Kafka read limits.
• Avoid data skew by choosing effective partition keys.
• Improve sink throughput or use asynchronous and batched writes.
• Monitor broker disk, network, CPU, replication status, Spark processing time, and consumer lag.

Scaling only one stage may simply move the bottleneck to another stage, so the entire pipeline must be measured.

Offset tracking identifies the Kafka records that a streaming query has consumed. Spark tracks offsets for each topic partition so that input progress can be reconstructed.

Checkpointing stores recovery metadata in durable storage. Depending on the query, it may contain:

• Processed source offsets
• Commit information
• Query metadata
• State-store data for stateful operations
• Information required to reconstruct the streaming execution

Recovery process:

1. The Spark application fails or is stopped.
2. It is restarted using the same query and checkpoint directory.
3. Spark reads the checkpoint metadata.
4. It determines the Kafka offsets from which processing should continue.
5. Stateful operators restore their previous state where applicable.

Important considerations:

• The checkpoint directory should be stored on durable, distributed storage.
• Different streaming queries should not share the same checkpoint directory.
• Deleting checkpoints can cause the query to restart according to new offset settings and can lose processing state.
• Checkpointing supports fault recovery, but end-to-end exactly-once results also depend on the behavior of the output sink.

A production pipeline should use coordinated security, monitoring, and reliability controls.

Security practices:

• Use TLS to encrypt broker-client and broker-broker communication.
• Use supported authentication mechanisms such as SASL.
• Apply access control lists so producers and consumers can access only authorized topics and groups.
• Protect credentials using a secret-management system.
• Restrict network access and avoid exposing brokers directly to untrusted networks.

Monitoring practices:

• Monitor broker availability, disk usage, request latency, and network traffic.
• Track under-replicated or offline partitions.
• Measure consumer lag by topic, partition, and consumer group.
• Monitor Spark input rate, processing rate, batch duration, executor failures, and state-store size.
• Configure alerts for sustained lag, broker failures, and checkpoint errors.

Reliability practices:

• Use an appropriate replication factor and min.insync.replicas value.
• Configure producer acknowledgements, retries, and idempotence.
• Use durable Spark checkpoints.
• Design sinks to be transactional or idempotent where possible.
• Use dead-letter topics for malformed or repeatedly failing records.
• Test broker, executor, network, and sink failures.
• Plan topic retention and capacity so temporary consumer outages do not cause required data to expire.