Reading from Kafka with Spark in Apache Spark - Step-by-Step Execution

Concept Flow - Reading from Kafka with Spark

Start Spark Session

↓

Set Kafka Configurations

↓

Create Spark DataFrame from Kafka

↓

Select and Cast Kafka Data

↓

Process or Show Data

↓

Stop Spark Session

This flow shows how Spark connects to Kafka, reads data, processes it, and then stops.

Execution Sample

Apache Spark

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

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

selected_df = kafka_df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
selected_df.show()

spark.stop()

This code connects Spark to Kafka, reads messages from 'test-topic', and shows key and value as strings.

Execution Table

Step	Action	Spark Variable	Result/Output
1	Start Spark Session	spark	SparkSession object created
2	Configure Kafka source	kafka_df	DataFrame configured to read from Kafka topic 'test-topic'
3	Load data from Kafka	kafka_df	DataFrame loaded with Kafka messages (key, value, topic, partition, offset, timestamp)
4	Select and cast key and value	selected_df	DataFrame with key and value as strings
5	Show data	Output	[Shows rows with key and value columns as strings]
6	Stop Spark Session	spark	SparkSession stopped
Exit	End of process		No more data to read or process

💡 Spark session stopped, no further Kafka data read

Variable Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	After Step 5	Final
spark	None	SparkSession created	SparkSession active	SparkSession active	SparkSession active	Stopped
kafka_df	None	Configured	Loaded with Kafka data	Loaded with Kafka data	Loaded with Kafka data	Released
selected_df	None	None	None	DataFrame with casted key/value	Same as previous	Released

Key Moments - 3 Insights

Why do we need to cast the key and value from Kafka data?

Kafka data comes as binary by default. Casting to string makes it readable and usable, as shown in step 4 of the execution_table.

What happens if the Kafka topic does not exist or is unreachable?

Spark will throw an error or hang at the load step (step 3), because it cannot connect or find the topic to read data.

Why do we stop the Spark session at the end?

Stopping the Spark session (step 6) frees resources and ends the streaming or batch job cleanly.

Visual Quiz - 3 Questions

Look at the execution_table, what is the state of 'kafka_df' after step 3?

A Configured but not loaded

B Loaded with Kafka data

C Stopped

D None

💡 Hint

Check the 'kafka_df' column in the execution_table row for step 3

At which step do we convert Kafka binary data to readable strings?

A Step 2

B Step 3

C Step 4

D Step 5

💡 Hint

Look for 'Select and cast key and value' in the execution_table

If we do not stop the Spark session, what is the likely outcome?

A Spark session continues running and holds resources

B Spark session automatically stops

C Kafka stops sending data

D Data is lost

💡 Hint

Refer to the key_moments explanation about stopping Spark session

Concept Snapshot

Reading from Kafka with Spark:
- Start SparkSession
- Configure Kafka source with bootstrap servers and topic
- Load Kafka data as DataFrame
- Cast key and value from binary to string
- Process or show data
- Stop SparkSession to release resources

Full Transcript

This visual execution shows how to read data from Kafka using Apache Spark. First, we start a Spark session. Then, we configure the Kafka source by specifying the Kafka servers and the topic to subscribe to. Next, we load the data from Kafka into a Spark DataFrame. Since Kafka data is in binary format, we cast the key and value columns to strings to make them readable. After that, we can process or display the data. Finally, we stop the Spark session to free resources. The execution table traces each step and variable state, helping beginners understand the flow and transformations.

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