Integrating Kafka with PySpark

In this blog we are going to discuss about how to integrate Apache Kafka with Spark using Python and its required configuration.

How Kafka works ?

Kafka is a distributed system consisting of servers and clients that communicate via a high-performance TCP network protocol. It can be deployed on bare-metal hardware, virtual machines, and containers in on-premise as well as cloud environments.

Topics

Brokers

Producers

Consumers

Role of Zookeeper in Kafka

Recap

Version compatibility to integrate Kafka with Spark

A python version with Kafka is compatible with version above 2.7

In order to integrate Kafka with Spark we need to use spark-streaming-kafka packages. The below are the version available for this packages.

It clearly shows that in spark-streaming-kafka-0–10 version the Direct Dstream is available. Using this version we can fetch the data in the form of Data Frame directly.

Therefore in order to run the structured streaming we need to use below mentioned package.

/usr/bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0 pyspark_structured_streaming_kafka_demo.py

Let us see how we decide the package because if we don’t choose the correct package we might face compatibility issues.

-- packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.3.0.

In this package, 0–10 refers to the spark-streaming-kafka version. If we choose to use structured streaming go with 0–10 version and if we choose to go with createStream functions we need to choose 0–8 version.

2.11 refers to the Scala version and 2.3.0 refers to the spark version.

Now based on your configuration choose the corresponding package, otherwise we will face the compatibility issues.

Integrating Kafka with Spark using Python sample code

Kafka Sample commands

Starting the server:

zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

kafka-server-start /usr/local/etc/kafka/server.properties

Topic creating and its use:

kafka-topics --zookeeper localhost:2181 --create --replication-factor 1 --partitions 1 --topic test

kafka-topic --zookeeper localhost:2181 --list

kafka-topic --zookeeper localhost:2181 --topic test --describe

kafka-topic --zookeeper localhost:2181 --topic second --delete

kafka-console-producer --bootstrap-server localhost:9092 --topic test

kafka-console-consumer --bootstrap-server localhost:9092 -- topic test

Producing Data using Python

Consuming Data using Python

Spark code for integration with Kafka

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import math
import string
import random

KAFKA_INPUT_TOPIC_NAME_CONS = “inputmallstream”
KAFKA_OUTPUT_TOPIC_NAME_CONS = “outputmallstream”
KAFKA_BOOTSTRAP_SERVERS_CONS = ‘localhost:9092’
MALL_LONGITUDE=78.446841
MALL_LATITUDE=17.427229
MALL_THRESHOLD_DISTANCE=100

if __name__ == “__main__”:

print(“PySpark Structured Streaming with Kafka Application Started …”)

spark = SparkSession \
.builder \
.appName(“PySpark Structured Streaming with Kafka”) \
.master(“local[*]”) \
.getOrCreate()

# One way of having the customer data in a table
# customer_data_df=spark.sql(“SELECT * FROM SPA_ASSGN_WAREHOUSE.MYMALL_CUSTOMER_DETAILS”)
# Other way is to create a dataframe from the csv file placed in HDFS location

customer_data_df=spark.read.format(“csv”).option(“header”,”true”).option(“delimiter”,”,”).load(“/home/cloudera/workspace/projects/SPAAssignment/data/customer.csv”)

print(“Printing Schema of customer_data_df: “)
customer_data_df.printSchema()

# Construct a streaming DataFrame that reads from testtopic
stream_detail_df = spark \
.readStream \
.format(“kafka”) \
.option(“kafka.bootstrap.servers”, KAFKA_BOOTSTRAP_SERVERS_CONS) \
.option(“subscribe”, KAFKA_INPUT_TOPIC_NAME_CONS) \
.option(“startingOffsets”, “latest”) \
.load()

stream_detail_df = stream_detail_df.selectExpr(“CAST(key AS STRING)”,”CAST(value AS STRING)”, “timestamp”)

split_col = split(stream_detail_df[‘value’], ‘,’)

stream_detail_df = stream_detail_df.withColumn(‘Cust_id’, split_col.getItem(0))

stream_detail_df = stream_detail_df.withColumn(‘IP address’, split_col.getItem(1))

stream_detail_df = stream_detail_df.withColumn(‘Latitude’, split_col.getItem(2).cast(“float”))

stream_detail_df = stream_detail_df.withColumn(‘Longitude’, split_col.getItem(3).cast(“float”))

print(“Printing Schema of stream_detail_df: “)
stream_detail_df.printSchema()

#Obtaining distance in meters using longitude and latitude

stream_detail_df = stream_detail_df.withColumn(‘a’, (
pow(sin(radians(col(“Latitude”) — lit(MALL_LATITUDE)) / 2), 2) +
cos(radians(lit(MALL_LATITUDE))) * cos(radians(col(“Latitude”))) *
pow(sin(radians(col(“Longitude”) — lit(MALL_LONGITUDE)) / 2), 2)
)).withColumn(“distance”, atan2(sqrt(col(“a”)), sqrt(-col(“a”) + 1)) * 12742000)

#Filtering customers based on distance

stream_detail_df = stream_detail_df.drop(“a”)

stream_detail_df = stream_detail_df.filter(col(“distance”) <= MALL_THRESHOLD_DISTANCE)

#Joining Customer stream data with customer dataset

stream_detail_df = stream_detail_df.join(customer_data_df,stream_detail_df.Cust_id == customer_data_df.CustomerID)

#Discount and Coupon generation

stream_detail_df = stream_detail_df.withColumn(“Discount”, when( (col(“Spending Score”) >= 85) | ((col(“Annual Income”) >=30000) & (col(“Spending Score”) >= 65)) , ‘40%’).when(((col(“Spending Score”) >= 65 ) & ( col(“Spending Score”) < 85 )) | ((col(“Annual Income”) >=20000) & (col(“Annual Income”) < 30000) & (col(“Spending Score”) >=45)), ‘30%’).when(((col(“Spending Score”) >= 45 ) & ( col(“Spending Score”) < 65)) | (col(“Annual Income”) >= 15000), ‘20%’).otherwise(“10%”))

stream_detail_df = stream_detail_df.withColumn(“Coupon_Code”, lit(‘’.join(random.choice(string.ascii_uppercase+string.ascii_lowercase+string.digits) for _ in range(16))))

# Write final result into console for debugging purpose
customer_detail_write_stream = stream_detail_df \
.writeStream \
.trigger(processingTime=’1 seconds’) \
.outputMode(“update”) \
.option(“truncate”, “false”)\
.format(“console”) \
.start()

# Output topic dataframe creation by selecting required columns

final_stream_df = stream_detail_df.selectExpr(“CustomerID”,”distance”,”Discount”,”Coupon_Code”)

final_stream_df = final_stream_df.withColumn(“key”,rand()*3)

# Write key-value data from a DataFrame to a specific Kafka topic specified in an option

customer_detail_write_stream_1 = final_stream_df \
.selectExpr(“CAST(key AS STRING)”, “to_json(struct(*)) AS value”) \
.writeStream \
.format(“kafka”) \
.option(“kafka.bootstrap.servers”, KAFKA_BOOTSTRAP_SERVERS_CONS) \
.option(“topic”, KAFKA_OUTPUT_TOPIC_NAME_CONS) \
.trigger(processingTime=’1 seconds’) \
.outputMode(“update”) \
.option(“checkpointLocation”, “/home/cloudera/workspace/SPAAssignment/chkpoint”) \
.start()

customer_detail_write_stream.awaitTermination()

print(“PySpark Structured Streaming with Kafka Application Completed.”)

Hope you liked this blog. Happy Learning!!!!!