Pyspark Strectured Streaming. awaittermination () вызывает java.lang.noclassdeffounderror: scala/serializable ⇐ Python

[Anonymous]

Скрипт Python ниже считывает данные JSON из темы Kafka с использованием структурированной потоковой передачи Spark. Любое действие после записи, связанное с этим рамком данных, вызывает ошибку в Spark3-conenct.log
MicroBatchEcecution: тест запроса [ID = D3856D89-78F2-41E7-84F1-635C1AD1B831, RunID = 4E301D2E-56DE-47AB-IFA210310333333131333331]. Ошибка
Java.lang.noclassDeffounderror: Scala /Serializable < /p>
Что может быть причиной этой ошибки? /> < /ul>
Вот пример < /p>

Код: Выделить всё

from pyspark.sql.types import StructType, StringType, StructField, IntegerType
from pyspark.sql import SparkSession
from pyspark.sql.connect.functions import col, from_json

if __name__ == "__main__":
spark = SparkSession.builder.remote("sc://hadoop:15002") \
.appName('SparkKafkaJsonToAvro') \
.config("packages", "org.apache.spark:spark-sql-kafka-0-10_2.13:3.5.2") \
.getOrCreate()

json_schema = StructType([
StructField("airline", StringType()),
StructField("destination", StructType([
StructField("airport", StringType()),
StructField("city", StringType()),
StructField("country", StringType()),
StructField("iata", StringType()),
StructField("icao", StringType()),
StructField("state", StringType())
])),
StructField("origin", StructType([
StructField("airport", StringType()),
StructField("city", StringType()),
StructField("country", StringType()),
StructField("iata", StringType()),
StructField("icao", StringType()),
StructField("state", StringType())
])),
StructField("price", IntegerType()),
StructField("stops", IntegerType())
])

df = spark.readStream.format('kafka') \
.option("kafka.bootstrap.servers", "kafka:9092") \
.option("startingOffsets", "earliest") \
.option("subscribe", "flights-json") \
.load()

df_json = df.select(from_json(col("value").cast("string"), json_schema).alias("value"))

query = df_json.writeStream.format("console").outputMode("append").start()
print(query.status)
query.awaitTermination()
< /code>
и его вывод в pycharm < /p>
{'message': 'Terminated with exception: scala/Serializable', 'isDataAvailable': False, 'isTriggerActive': False}
pyspark.errors.exceptions.connect.StreamingQueryException: [STREAM_FAILED] Query [id = 9b9938a6-90b9-44f8-9bb8-de9b0ebe433e, runId = c8a7128e-a770-40f2-9801-ef635795bb39] terminated with exception: scala/Serializable

Я попытался изменить формат записи на kafka, память, но статус и awaittermination () Метод возврата
Я ожидаю>

Подробнее здесь: https://stackoverflow.com/questions/795 ... lassdeffou