如何使用Java中的结构化流来从Kafka反序列化记录？

我使用Spark 2.1.

我试图使用Spark Structured Streaming从Kafka读取记录,反序列化它们并在之后应用聚合.

我有以下代码:

SparkSession spark = SparkSession
        .builder()
        .appName("Statistics")
        .getOrCreate();

Dataset df = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", kafkaUri)
        .option("subscribe", "Statistics")
        .option("startingOffsets", "earliest")
        .load();

df.selectExpr("CAST(value AS STRING)")

public StatisticsRecord deserialize(String s, byte[] bytes)