[Python] Spark job stuck at running process

I use docker-compose to make a cluster of Spark and Kafka. When I tried to consume a topic by submitting it to this cluster, the job was always stuck and not executed.

Here is my docker-compose file:

version: '3'
services:
spark-master:
image: spark-cluster:1.0
container_name: spark-master
hostname: spark-master
environment:
- SPARK_MODE=master
- SPARK_MASTER_HOSTNAME=spark-master
- SPARK_MASTER_PORT=7077
ports:
- "4040:4040"
- "6066:6066"
- "7077:7077"
- "8080:8080"
deploy:
resources:
limits:
cpus: "0.5" # Adjust as needed
memory: 256M # Adjust as needed
# command: bin/spark-class org.apache.spark.deploy.master.Master
networks:
- spark-network

spark-worker-1:
image: spark-cluster:1.0
container_name: spark-worker-1
environment:
- SPARK_MODE=worker
- SPARK_MASTER_URL=spark://spark-master:7077
- SPARK_WORKER_MEMORY = 4g
deploy:
resources:
limits:
cpus: "4" # Adjust as needed
memory: 5G # Adjust as needed
depends_on:
- spark-master
# command: bin/spark-class org.apache.spark.deploy.worker.Worker
networks:
- spark-network

zookeeper:
image: bitnami/zookeeper:3.9
container_name: zookeeper-server
restart: always
ports:
- "2181:2181"
environment:
- ALLOW_ANONYMOUS_LOGIN=yes

kafka1:
image: bitnami/kafka:3.5
container_name: broker-1
ports:
- "9093:9093"
environment:
- KAFKA_BROKER_ID=1
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9093,INTERNAL://:9092
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9093,INTERNAL://:9092
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXTLAINTEXT,INTERNALLAINTEXT
- KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
depends_on:
- zookeeper

kafka2:
image: bitnami/kafka:3.5
container_name: broker-2
ports:
- "9094:9094"
environment:
- KAFKA_BROKER_ID=2
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9094,INTERNAL://:9092
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9094,INTERNAL://:9092
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXTLAINTEXT,INTERNALLAINTEXT
- KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
depends_on:
- zookeeper

kafka3:
image: bitnami/kafka:3.5
container_name: broker-3
ports:
- "9095:9095"
environment:
- KAFKA_BROKER_ID=3
- KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9095,INTERNAL://:9092
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9095,INTERNAL://:9092
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXTLAINTEXT,INTERNALLAINTEXT
- KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
depends_on:
- zookeeper

networks:
spark-network:
driver: bridge


and my consumer.py:

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
import os
scala_version = '2.12'
spark_version = '3.5.0'
packages = [
f'org.apache.spark:spark-sql-kafka-0-10_{scala_version}:{spark_version}',
'org.apache.kafka:kafka-clients:3.5.0',
'org.apache.hadoop:hadoop-client:3.0.0',
]

spark = SparkSession.builder \
.master("spark://172.18.32.1:7077") \
.appName("kafka-example") \
.config("spark.jars.packages", ",".join(packages)) \
.getOrCreate()

spark.sparkContext.setLogLevel("ERROR")
# Địa chỉ của Kafka broker
# kafka_brokers = "localhost:9092"
kafka_brokers = "localhost:9095"

# Định nghĩa Kafka topic bạn muốn đọc dữ liệu
kafka_topic = "mytopic1"

# Đọc dữ liệu từ Kafka
df = spark \
.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_brokers) \
.option("subscribe", kafka_topic) \
.option("startingOffsets", "earliest") \
.load()

# Hiển thị dữ liệu từ Kafka
castDf = df .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
query = castDf.writeStream \
.outputMode("append") \
.format("console") \
.start()

query.awaitTermination()


The output when I run this code:

Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
23/11/28 23:08:17 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.

[Stage 0:> (0 + 0) / 1]


And I look at the Job on UI:
Master UI
However, when I submitted this code to standalone cluster, it worked fine.

Can sb help me, please? Sorry for my bad English.

I hope for a solution to this problem. By the way, the image "spark-cluster:1.0" I built from "bitnami/spark:3.5"

Continue reading...