[Python] Unable to read messages from kafka broker with PySpark

Problem: I am trying to connect to a Kafka broker using PySpark, but when consuming messages from the test-topic, I receive empty NULL values instead of the expected JSON content.

consumer.py code

def consume_message():

load_dotenv()

spark = SparkSession.builder \
.appName("KafkaReader") \
.config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
.getOrCreate()

schema = StructType().add("message", StringType())

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

messages = df.select(from_json(col("value").cast("string"), schema).alias("data")) \
.select("data.message")

query = messages \
.writeStream \
.outputMode("append") \
.format("console") \
.start()
query.awaitTermination()


if __name__ == '__main__':
time.sleep(10)
while True:
consume_message()
time.sleep(60)


When I run the consumer script, the console output is as follows, with empty NULL values for the message:

2025-03-05 23:15:16 -------------------------------------------
2025-03-05 23:15:16 Batch: 25
2025-03-05 23:15:16 -------------------------------------------
2025-03-05 23:15:16 +-------+
2025-03-05 23:15:16 |message|
2025-03-05 23:15:16 +-------+
2025-03-05 23:15:16 | NULL|
2025-03-05 23:15:16 +-------+
2025-03-05 23:15:16
2025-03-05 23:16:18 -------------------------------------------
2025-03-05 23:16:18 Batch: 26
2025-03-05 23:16:18 -------------------------------------------
2025-03-05 23:16:18 +-------+
2025-03-05 23:16:18 |message|
2025-03-05 23:16:18 +-------+
2025-03-05 23:16:18 | NULL|
2025-03-05 23:16:18 +-------+


When accessing the Kafka broker directly using the kafka-console-consumer.sh tool, I see the correct JSON messages:

docker exec -it <container_id> bash

cd opt
cd bitnami
cd kafka
cd bin

kafka-console-consumer.sh --bootstrap-server kafka:9093 --topic test-topic --from-begin
ning

{"date": "2025-03-05 22:03", "BTCUSDT": "90244.50000000", "ETHUSDT": "2230.69000000", "BNBUSDT": "595.95000000", "SOLUSDT": "145.04000000", "ADAUSDT": "0.98140000", "XRPUSDT": "2.49800000", "SHIBUSDT": "0.00001328", "DOGEUSDT": "0.20369000", "TONUSDT": "3.02600000"}
{"date": "2025-03-05 22:04", "BTCUSDT": "90233.10000000", "ETHUSDT": "2230.99000000", "BNBUSDT": "595.91000000", "SOLUSDT": "145.02000000", "ADAUSDT": "0.98330000", "XRPUSDT": "2.49890000", "SHIBUSDT": "0.00001330", "DOGEUSDT": "0.20379000", "TONUSDT": "3.02600000"}


Tried changing schema, but result is the same

schema = StructType([StructField("message", StringType())])

Continue reading...