Apache Kafka is a widely used distributed streaming platform that specializes in providing a high-throughput, fault-tolerant method for managing data feeds. Apache Spark, on the other hand, is an open-source unified analytics engine for large-scale data processing. Integrating Kafka with Spark enables the processing of real-time data streams in addition to batch data processing. This article focuses on how to read data from a Kafka topic in a Spark batch job, an essential capability for businesses that need to process historical data stored in Kafka.

Understanding Kafka and Spark Integration

Spark provides a Kafka integration module, spark-sql-kafka-0-10, which supports reading from and writing to Kafka. The integration allows Spark to process data directly from Kafka topics both in streaming and batch modes. The primary class responsible for this integration is DataFrameReader for reading operations.

Setting Up Spark with Kafka

To begin with, ensure that your Spark cluster has the necessary Kafka dependencies. The Kafka integration package can be included in your Spark application using the --packages option during the submission of your Spark job:

./bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.1.1 ...

Reading from a Kafka Topic in a Batch Job

Reading from a Kafka topic in a batch job involves setting up a DataFrameReader to load data from Kafka as a DataFrame. Here is an example code snippet using PySpark, the Python API for Spark:

1from pyspark.sql import SparkSession
2
3# Create Spark session
4spark = SparkSession.builder \
5    .appName("KafkaReadBatch") \
6    .getOrCreate()
7
8# Read data from Kafka
9df = spark \
10  .read \
11  .format("kafka") \
12  .option("kafka.bootstrap.servers", "localhost:9092") \
13  .option("subscribe", "topic1") \
14  .option("startingOffsets", "earliest") \
15  .option("endingOffsets", "latest") \
16  .load()
17
18# Show dataframe
19df.show()

In the above example:

• kafka.bootstrap.servers specifies the Kafka server's address.
• subscribe sets the Kafka topic from which to read.
• startingOffsets and endingOffsets define the range of data to read. In this case, it's from the very beginning to the most recent data available, making it suitable for batch processing.

Key Configuration Options

• startingOffsets: Can be either earliest, latest (not useful in batch scenarios), or a JSON string specifying offsets.
• endingOffsets: Can be latest, or a JSON string specifying the last offset to fetch.

Processing Data

Once the data is read into a DataFrame, it can be processed using Spark’s DataFrame API. For instance, you may want to decode the binary Kafka records into string format if they are simple text messages:

1from pyspark.sql.functions import col, expr
2
3# Decode the Kafka record
4decoded_df = df.select(
5    expr("cast (value as string) as actualValue")
6)
7decoded_df.show()

Summary Table

Conclusion

This method enables batch processing of historical Kafka topic data, which is beneficial for analytical and reporting purposes. The seamless integration between Kafka and Spark allows businesses to leverage both streaming and batch data processing capabilities to make more data-driven decisions.