Setting Up Apache Kafka

Before diving into the coding of the Kafka producer and consumer, it's essential to establish an Apache Kafka cluster. For comprehensive guidance on setting up Kafka either locally or through a cloud service, refer to the official Apache Kafka documentation.

Creating a Kafka Producer with Python

To initiate, we will create a simple Python script (producer.py) that generates JSON data and transmits it to a Kafka topic. We'll utilize the kafka-python library, which is a widely-used Python client for Kafka.

from kafka import KafkaProducer

import json

import time

import random

# Kafka broker

bootstrap_servers = ['localhost:9092']

# Initialize Kafka producer

producer = KafkaProducer(bootstrap_servers=bootstrap_servers,

value_serializer=lambda v: json.dumps(v).encode('utf-8'))

# Generate and send sample data

Names = ["Khayyon", "Parker", "Keith", "Tyler", "Nyla", "Brian", "Georgia", "Bry", "Nu"]

for i in range(100):

data = {"Name": random.choice(Names), "Age": i, "Gender": "Male" if i % 2 == 0 else "Female"}

producer.send('test', value=data)

time.sleep(1)

# Close producer

producer.close()

The script begins by initializing a Kafka producer, which requires the server's IP address and port number, typically localhost:9092 for local installations. A lambda function is provided for the value_serializer, which converts the data into a format that Kafka can process. Essentially, the value_serializer is responsible for transforming structured data, such as JSON objects, into bytes that Kafka can store and transmit, a process known as serialization. This script will populate the Kafka topic with 100 entries, generating random names and alternating genders for each record, simulating real-time data flow for our consumer script.

Building a Kafka Consumer with PySpark

Next, we will create a PySpark script (consumer.py) to consume data from the Kafka topic and process it. PySpark integrates smoothly with Kafka through the spark-sql-kafka library, enabling us to read streaming data from Kafka topics into Spark DataFrames.

from pyspark.sql import SparkSession

from pyspark.sql.types import StructType, StringType, IntegerType

from pyspark.sql.functions import from_json, col

# Initialize SparkSession

spark = SparkSession.builder

.appName("KafkaConsumer")

.getOrCreate()

# Stream data from Kafka Topic

df = spark.readStream

.format("kafka")

.option("kafka.bootstrap.servers", "localhost:9092")

.option("subscribe", "test")

.load()

# Convert Kafka value from binary to string

kafka_df = df.selectExpr("CAST(value AS STRING)")

# Output the DataFrame to the console

query = kafka_df

.writeStream

.outputMode("append")

.format("console")

.start()

# Await termination of the streaming query

query.awaitTermination()

# Stop SparkSession

spark.stop()

In this consumer script, we initialize a SparkSession, read data from the Kafka topic into a DataFrame, and then print that DataFrame to the console. Like the producer script, it requires the Kafka broker and topic details. This script will display the data in real time as it is generated by the producer, allowing us to observe the results interactively.

Integrating Kafka with PySpark

With both the Kafka producer and consumer scripts complete, we can now integrate Kafka messaging with PySpark. Begin by executing the Kafka producer to start generating JSON data for the Kafka topic. Next, run the PySpark consumer script to process that data in real time.

The following video demonstrates building a streaming solution using Kafka, PySpark, and Apache HUDI with practical code examples.

This video provides an additional perspective on Spark Structured Streaming with Kafka, showcasing a hands-on use case.