Kafka protobuf 的 Spark 结构化流答案

【问题标题】：Spark structured streaming of Kafka protobufKafka protobuf 的 Spark 结构化流
【发布时间】：2020-01-31 22:06:38
【问题描述】：

我正在尝试创建一个使用 ProtoBuf 编码的 Kafka 消息的 Spark Streaming。

这是我最近几天尝试的方法

    import spark.implicits._
    def parseLine (str: Array[Byte]): ProtoSchema = ProtoSchema.parseFrom(str)   
    val storageLoc: String = "/tmp/avl/output"
    val checkpointLoc: String = "/tmp/avl/checkpoint"
    val dfStreamReader: DataFrame = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokers)
      .option("failOnDataLoss", value = false)
      .option("subscribe", topics)
      .load()

    val dfStreamReaderValues: Dataset[Array[Byte]] = dfStreamReader.map(row => row.getAs[Array[Byte]]("value"))

    val rddProtoSchema: Dataset[ProtoSchema] = dfStreamReaderValues.map(str => parseLine(str))

    val dfRaw: DataFrame = spark.sqlContext.protoToDataFrame(rddProtoSchema.rdd)

    val streamWriterAirline: StreamingQuery = dfRaw.writeStream
      .format("parquet")
      .option("path", storageLoc)
      .option("checkpointLocation", checkpointLoc)
      .outputMode(Append)
      .trigger(ProcessingTime("2 seconds"))
      .start()
    spark.streams.awaitAnyTermination(20000)

使用 scalapb，我设法解码二进制 proto 文件并转换为数据帧。但是对于流式传输，我在编译时在解析行中得到了这个异常：

    val rddProtoSchema: Dataset[ProtoSchema] = dfStreamReaderValues.map(str => parseLine(str))
    >>>>>
    scala.ScalaReflectionException: <none> is not a term

谁能给点提示？

【问题讨论】：

标签： scala apache-spark protocol-buffers spark-streaming scalapb

【解决方案1】：

更新：sparksql-scalapb 现在能够为协议缓冲区派生编码器，并且不再需要以前使用 UDT 生成器的方法。说明是available here。

旧答案（现在不相关）：使用数据集时，Spark 会尝试为消息中的每个字段查找 SQL 类型。 Spark 不知道如何处理 ScalaPB 枚举（它们被表示为由 case 对象扩展的密封特征），因此它给出了这个错误。解决方法是注册枚举和用户定义的类型。这可以按如下方式完成：

将 sparksql-scalapb-gen 添加到您的 project/plugins.sbt（而不是您的主要 build.sbt）：

libraryDependencies += "com.thesamet.scalapb" %% "sparksql-scalapb-gen" % "0.8.1"

检查以上版本是否与您正在使用的sparksql-scalapb 的版本匹配。

将此生成器添加到您的PB.targetsbuild.sbt：

PB.targets in Compile := Seq(
  scalapb.gen() -> (sourceManaged in Compile).value,
  scalapb.UdtGenerator -> (sourceManaged in Compile).value
)

重新生成源（可能需要sbt clean，后跟sbt compile）
在主函数中调用生成的注册函数。应该是mypackage.MyFileUdt.register()

见：https://scalapb.github.io/sparksql.html#datasets-and-none-is-not-a-term

示例项目：https://github.com/thesamet/sparksql-scalapb-test

【讨论】：