Я прочитал файл csv, преобразовал поле значения в байты и написал в тему Kafka, используя приложение производителя Kafka. Теперь я пытаюсь читать из темы Kafka, используя структурированную потоковую передачу, но не могу применить настраиваемую десериализацию kryo в поле значения.
Может ли кто-нибудь сказать мне, как использовать пользовательскую десериализацию в структурированной потоковой передаче?
У меня была аналогичная проблема, в основном, у меня были все сообщения Kafka на Protobuf, и я решил это с помощью UDF.
from pyspark.sql.functions import udf
def deserialization_function(message):
#You need to add your code to deserialize your messages
#I returned a json but you can return other structure
json = {"x": x_deserializable,
"y": y_deserializable,
"w": w_deserializable,
"z": z_deserializable,
return json
schema = StructType() \
.add("x", TimestampType()) \
.add("y", StringType()) \
.add("z", StringType()) \
.add("w", StringType())
own_udf = udf(deserialization_function, schema)
stream = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
.option("subscribe", topic) \
.load()
query = stream \
.select(col("value")) \
.select((own_udf("value")).alias("value_udf")) \
.select("value_udf.x", "value_udf.y", "value_udf.w", "value_udf.z")
