Spark结构化流-如何将字节值排队到Kafka？

Question

我正在编写一个使用结构化流的 Spark 应用程序。应用从 Kafka 主题主题 1 读取消息，构造新消息，将其序列化为 Array[Byte]，并将其发布到另一个 Kafka 主题主题 2。

字节数组的序列化很重要，因为我使用了topic2的下游消费者也使用的特定序列化器/反序列化器。

不过，我很难向Kafka生产。我甚至不知道该怎么做。网上只有很多关于排队JSON数据的例子。

代码-

case class OutputMessage(id: String, bytes: Array[Byte])

implicit val encoder: Encoder[OutputMessage] = org.apache.spark.sql.Encoders.kryo

val outputMessagesDataSet: DataSet[OutputMessage] = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "server1")
  .option("subscribe", "topic1")
  .load()
  .select($"value")
  .mapPartitions{r =>
     val messages: Iterator[OutputMessage] = createMessages(r)
     messages
  }

outputMessagesDataSet
  .writeStream
  .selectExpr("CAST(id AS String) AS key", "bytes AS value")
  .format("kafka")
  .option("kafka.bootstrap.servers", "server1")
  .option("topic", "topic2")
  .option("checkpointLocation", loc)
  .trigger(trigger)
  .start
  .awaitTermination

但是，这会引发异常org.apache.spark.sql.AnalysisException：无法解析给定输入列的“id”：[value]；第1行pos 5；

如何以id作为键并以bytes作为值排队到Kafka？

Answer 1

正如@ EmiCareOfCell44所建议的，我打印出了这个模式

如果我做消息数据集.printSchema（）， 那么我只得到一个二进制类型的值。但是如果我这样做

val df = spark.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "server1")
.option("subscribe", "topic1")
.load()

df.printSchema()

然后打印

 root
  |-- key: binary (nullable = true)
  |-- value: binary (nullable = true)
  |-- topic: string (nullable = true)
  |-- partition: integer (nullable = true)
  |-- offset: long (nullable = true)
  |-- timestamp: timestamp (nullable = true)
  |-- timestampType: integer (nullable = true)

但是数据帧没有经历所需的转换，这是在

.mapPartitions{r =>
 val messages: Iterator[OutputMessage] = createMessages(r)
 messages
}

看起来数据集的值只有一个二进制值。

我在这里搜索了一些答案，然后发现Spark Dataset mapGroups操作后这个后值类型是二进制的，甚至在函数中返回一个字符串

我设置了一个编码器-

implicit val encoder: Encoder[OutputMessage] = org.apache.spark.sql.Encoders.kryo

这导致该值转换为二进制。由于输出消息是一个 scala 类，编码器不是必需的，所以我删除了它。之后，打印出架构显示两个字段（字符串和字节，这是我想要的）。之后，行 .selectExpr（“CAST（ID 为字符串）为键”、“字节为值”）运行良好。

Answer 2

您可以检查“收集”消息的数据帧的架构。由于您仅收集“值”字段，因此传入事件将以以下形式到达：

    +-------------------+
    | value             |
    +-------------------+
    | field1,field2,..  |
    +-------------------+
  

Yo 还需要查询密钥，就像在 Spark 文档中一样：

df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

或

df.select(col("key").cast(StringType), col("value").cast(StringType))