Reading from Kafka with Spark in Apache Spark - Mini Project: Build & Apply

Reading from Kafka with Spark

📖 Scenario: You work at a company that collects real-time data from various sensors. This data is sent to a Kafka topic. Your job is to read this data using Apache Spark and prepare it for analysis.

🎯 Goal: Build a Spark application that reads messages from a Kafka topic, extracts the message values, and displays them.

📋 What You'll Learn

Create a Spark session named spark

Read data from Kafka topic sensor-data on Kafka server localhost:9092

Select only the value column from the Kafka stream and cast it to string

Show the first 5 messages from the stream

💡 Why This Matters

🌍 Real World

Many companies use Kafka to collect real-time data streams from devices, logs, or user activity. Spark helps process this data quickly for monitoring or analytics.

💼 Career

Data engineers and data scientists often need to read streaming data from Kafka using Spark to build real-time data pipelines and dashboards.

1

Create Spark session

Create a Spark session called spark with the app name KafkaSparkApp.

Apache Spark

# Create a Spark session named spark with app name 'KafkaSparkApp'
# Your code here

Hint

Use SparkSession.builder.appName(...).getOrCreate() to create the Spark session.

2

Configure Kafka source

Create a DataFrame called kafka_df by reading from Kafka with spark.read.format("kafka"). Set the Kafka server to localhost:9092 and the topic to sensor-data. Use load() to load the data.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("KafkaSparkApp").getOrCreate()

# Read from Kafka topic 'sensor-data' on server 'localhost:9092' into kafka_df
# Your code here

Hint

Use option to set Kafka server and topic before calling load().

3

Extract and cast message values

Create a new DataFrame called messages_df by selecting the value column from kafka_df and casting it to string using .cast("string").

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("KafkaSparkApp").getOrCreate()

kafka_df = spark.read.format("kafka")\
    .option("kafka.bootstrap.servers", "localhost:9092")\
    .option("subscribe", "sensor-data")\
    .load()

# Select the 'value' column from kafka_df and cast it to string into messages_df
# Your code here

Hint

Use selectExpr("CAST(value AS STRING) as value") to cast the Kafka message values.

4

Show messages

Use messages_df.show(5) to display the first 5 messages from the Kafka stream.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("KafkaSparkApp").getOrCreate()

kafka_df = spark.read.format("kafka")\
    .option("kafka.bootstrap.servers", "localhost:9092")\
    .option("subscribe", "sensor-data")\
    .load()

messages_df = kafka_df.selectExpr("CAST(value AS STRING) as value")

# Show the first 5 messages from messages_df
# Your code here

Hint

Use show(5) on messages_df to display the messages.

Practice

1. What is the main purpose of using spark.readStream.format("kafka") in Apache Spark?

easy

A. To read live streaming data from Kafka topics

B. To write data into Kafka topics

C. To stop Kafka streaming jobs

D. To configure Kafka brokers

Solution

1. Step 1: Understand the function of spark.readStream.format("kafka")

   This command sets up Spark to read streaming data from Kafka, meaning it connects Spark to Kafka to get live data.

2. Step 2: Differentiate reading from writing or configuring

   Writing data or stopping jobs are different operations. This command is specifically for reading data streams.

3. Final Answer:

   To read live streaming data from Kafka topics -> Option A

4. Quick Check:

   Reading Kafka streams = To read live streaming data from Kafka topics [OK]

Hint: Remember: readStream + kafka = reading live data [OK]

Common Mistakes:

• Confusing reading with writing data
• Thinking it configures Kafka brokers
• Assuming it stops streaming jobs

2. Which of the following is the correct way to specify Kafka bootstrap servers when reading from Kafka in Spark?

easy

A. option("servers", "localhost:9092")

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

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

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

Solution

1. Step 1: Recall the exact option key for Kafka bootstrap servers

   The correct key is "bootstrap.servers" without the "kafka." prefix when used in Spark's option method.

2. Step 2: Compare all options

   Only option("bootstrap.servers", "localhost:9092") uses "bootstrap.servers" correctly. Others use incorrect keys with unnecessary prefixes.

3. Final Answer:

   option("bootstrap.servers", "localhost:9092") -> Option D

4. Quick Check:

   Correct Kafka server option key = option("bootstrap.servers", "localhost:9092") [OK]

Hint: Use exact key 'bootstrap.servers' for Kafka servers [OK]

Common Mistakes:

• Adding 'kafka.' prefix to option keys
• Using incorrect option names like 'kafka.servers'
• Omitting required option keys

3. Given the following Spark code snippet reading from Kafka, what will be the schema of the resulting DataFrame?

df = spark.readStream.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("subscribe", "test-topic")\
  .load()

medium

A. DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int)

B. DataFrame with columns: key (string), value (string)

C. DataFrame with columns: message (string), offset (int)

D. DataFrame with columns: topic (string), message (string), timestamp (string)

Solution

1. Step 1: Understand default Kafka DataFrame schema in Spark

   When reading from Kafka, Spark returns a DataFrame with columns including key and value as binary types, plus metadata columns like topic, partition, offset, timestamp, and timestampType.

2. Step 2: Match schema with options

   DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) correctly lists all these columns with their types. Other options simplify or rename columns incorrectly.

3. Final Answer:

   DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) -> Option A

4. Quick Check:

   Kafka DataFrame schema = DataFrame with columns: key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (timestamp), timestampType (int) [OK]

Hint: Kafka data columns include key and value as binary by default [OK]

Common Mistakes:

• Assuming key and value are strings by default
• Ignoring metadata columns like offset and timestamp
• Confusing column names and types

4. Identify the error in the following Spark code reading from Kafka:

df = spark.readStream.format("kafka")\
  .option("kafka.bootstrap.servers", "localhost:9092")\
  .option("topic", "my-topic")\
  .load()

medium

A. The format should be 'kafkaStream' instead of 'kafka'

B. The option key for topic should be 'subscribe', not 'topic'

C. The bootstrap.servers option is misspelled

D. The load() method is missing parentheses

Solution

1. Step 1: Check the option keys for Kafka source in Spark

   The correct option key to specify Kafka topics to read is 'subscribe', not 'topic'.

2. Step 2: Verify other parts of the code

   Bootstrap servers key is correct, format 'kafka' is correct, and load() has parentheses.

3. Final Answer:

   The option key for topic should be 'subscribe', not 'topic' -> Option B

4. Quick Check:

   Use 'subscribe' to specify topics [OK]

Hint: Use 'subscribe' option for topics, not 'topic' [OK]

Common Mistakes:

• Using 'topic' instead of 'subscribe'
• Misspelling option keys
• Confusing format names

5. You want to read messages from Kafka in Spark and convert the binary 'value' column to a string column named 'message'. Which code snippet correctly achieves this?

hard

A. df.withColumn("message", df.value.toString())

B. df.select("value").cast("string").alias("message")

C. df.selectExpr("CAST(value AS STRING) as message")

D. df.select(df.value).withColumnRenamed("value", "message")

Solution

1. Step 1: Understand how to cast binary to string in Spark DataFrame

   The correct way is to use selectExpr with SQL expression: CAST(value AS STRING) and alias it as 'message'.

2. Step 2: Evaluate each option

   df.selectExpr("CAST(value AS STRING) as message") uses selectExpr correctly. df.select("value").cast("string").alias("message") tries to cast after select which is invalid. df.withColumn("message", df.value.toString()) uses toString() which converts to Python string, not Spark column. df.select(df.value).withColumnRenamed("value", "message") renames without casting to string.

3. Final Answer:

   df.selectExpr("CAST(value AS STRING) as message") -> Option C

4. Quick Check:

   Use selectExpr with CAST for binary to string [OK]

Hint: Use selectExpr("CAST(value AS STRING) as message") to convert [OK]

Common Mistakes:

• Using toString() on Spark columns
• Casting after select without aliasing
• Not renaming the column after casting