java - 如何使用 Spark 结构化流为 Kafka 流实现自定义反序列化器？

Question

我正在尝试将我当前基于使用 RDD(来自 their documentation )的流式应用程序迁移到他们使用结构化流式传输的新数据集 API，我被告知这是进行实时流式传输的首选方法这些天 Spark 。

目前我有一个应用程序设置来使用 1 个名为“SATELLITE”的主题，该主题包含包含关键时间戳的消息和包含 Satellite POJO 的值。但是我在弄清楚如何为此实现反序列化器时遇到了问题。在我当前的应用程序中，这很简单，您只需在您喜欢的 kafka 属性映射中添加一行 kafkaParams.put("value.deserializer", SatelliteMessageDeserializer.class);我在 Java 中执行此操作，这是最大的挑战，因为所有解决方案似乎都在 Scala 中，我不太了解它，而且我无法轻松地将 Scala 代码转换为 Java 代码。

我遵循了 this question 中概述的 JSON 示例，目前有效，但对于我需要做的事情来说似乎过于复杂。鉴于我已经为此目的制作了自定义反序列化器，我不明白为什么我必须先将其转换为字符串，只是将其转换为 JSON，然后再将其转换为我想要的类类型。我也一直在尝试使用我发现的一些示例 here ，但到目前为止我运气不好。

目前我的应用看起来像这样(使用 json 方法):

import common.model.Satellite;
import org.apache.spark.sql.*;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

public class SparkStructuredStreaming  implements Runnable{

    private String bootstrapServers;
    private SparkSession session;

    public SparkStructuredStreaming(final String bootstrapServers, final SparkSession session) {
        this.bootstrapServers = bootstrapServers;
        this.session = session;
    }
    @Override
    public void run() {
        Dataset<Row> df = session
                .readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", bootstrapServers)
                .option("subscribe", "SATELLITE")
                .load();

        StructType schema =  DataTypes.createStructType(new StructField[] {
                DataTypes.createStructField("id", DataTypes.StringType, true),
                DataTypes.createStructField("gms", DataTypes.StringType, true),
                DataTypes.createStructField("satelliteId", DataTypes.StringType, true),
                DataTypes.createStructField("signalId", DataTypes.StringType, true),
                DataTypes.createStructField("cnr", DataTypes.DoubleType, true),
                DataTypes.createStructField("constellation", DataTypes.StringType, true),
                DataTypes.createStructField("timestamp", DataTypes.TimestampType, true),
                DataTypes.createStructField("mountPoint", DataTypes.StringType, true),
                DataTypes.createStructField("pseudorange", DataTypes.DoubleType, true),
                DataTypes.createStructField("epochTime", DataTypes.IntegerType, true)
        });

            Dataset<Satellite> df1 = df.selectExpr("CAST(value AS STRING) as message")
                    .select(functions.from_json(functions.col("message"),schema).as("json"))
                    .select("json.*")
                    .as(Encoders.bean(Satellite.class));

        try {
            df1.writeStream()
                    .format("console")
                    .option("truncate", "false")
                    .start()
                    .awaitTermination();

        } catch (StreamingQueryException e) {
            e.printStackTrace();
        }
    }
}

我有一个看起来像这样的自定义反序列化器

import common.model.Satellite;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class SatelliteMessageDeserializer implements Deserializer<Satellite> {

    private static Logger logger = LoggerFactory.getLogger(SatelliteMessageDeserializer.class);
    private ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map configs, boolean isKey) {
    }

    @Override
    public void close() {
    }

    @Override
    public Satellite deserialize(String topic, byte[] data) {
        try {
            return objectMapper.readValue(new String(data, "UTF-8"), getMessageClass());
        } catch (Exception e) {
            logger.error("Unable to deserialize message {}", data, e);
            return null;
        }
    }

    protected Class<Satellite> getMessageClass() {
        return Satellite.class;
    }
}

如何在 SparkStructuredStreaming 类中使用我的自定义解串器？我正在使用 Spark 2.4、OpenJDK 10 和 Kafka 2.0

编辑:我已经尝试创建自己的 UDF，我认为这是应该如何完成的，但我不确定如何让它返回特定类型，因为它似乎只允许我使用 Datatypes 类中的那些!

UserDefinedFunction mode = udf(
                (byte[] bytes) -> deserializer.deserialize("", bytes), DataTypes.BinaryType //Needs to be type Satellite, but only allows ones of type DataTypes
        );

Dataset df1 = df.select(mode.apply(col("value")));

Answer 1

from_json 只能用于字符串类型的列。

Structured Streaming always consumes the Kafka values as bytes

Values are always deserialized as byte arrays with ByteArrayDeserializer. Use DataFrame operations to explicitly deserialize the values

因此，您首先至少要反序列化为一个字符串，但我认为您真的不需要这样做。

也许可以这样做

df.select(value).as(Encoders.bean(Satellite.class))

如果这不起作用，您可以尝试定义自己的 UDF/解码器，这样您就可以拥有类似 SATELLITE_DECODE(value)

的内容

在标度中

object SatelliteDeserializerWrapper {
    val deser = new SatelliteDeserializer
}
spark.udf.register("SATELLITE_DECODE", (topic: String, bytes: Array[Byte]) => 
    SatelliteDeserializerWrapper.deser.deserialize(topic, bytes)
)

df.selectExpr("""SATELLITE_DECODE("topic1", value) AS message""")

参见 this post for inspiration , 并且还提到了 in Databricks blog