使用Spark 2.0.2、Kafka源和calapb进行结构化流

Question

我使用结构化流媒体（Spark 2.0.2）来消费Kafka消息。使用scalapb，protobuf中的消息。我得到以下错误。请帮助。。

线程“main”scala中的异常。ScalaRefltionException：不是一个术语org.apache.spark.sql.catalyst.符号$SymbolApi$9.apply术语（Seflection.scala:592）org.apache.spark.sql.catalyst.符号$SymbolContextApiImpl.as术语（Symbols.scala:84）org.apache.spark.sql.catalyst.ScalaReflection$class.constructParams（ScalaReflection.scala:811）org.apache.spark.sql.catalyst.ScalaReflection$.构造参数（ScalaReflection.scala:39）org.apache.spark.sql.catalyst.ScalaReflection$class.getConstructorParameters（ScalaReflection.scala:800）org.apache.spark.sql.catalyst.ScalaReflection$. getConstructorParameters（ScalaReflection.scala:39）ScalaReflection$. org$apache$sql$催化剂$ScalaReflection$$序列化器对于（ScalaReflection.scala:582）org.apache.spark.sql.catalyst.ScalaReflection$. org$apache$火花$sql$催化剂$ScalaReflection$$序列化器对于（ScalaReflection.scala:460）scala.reflect.api.ScalaReflection$$anonfun$class.as（ScalaRymbols.scala:199）scala.reflect.internal.ScalaRScalaReflection$. org$apache$火花$sql$催化剂$ScalaReflection$$序列化器对于（ScalaReflection. scala： 583）在org. apache. sql.催化剂。ScalaReflection$.序列化器对于（ScalaReflection. scala： 425）在org. apache. sql.催化剂.编码器。ExpressionEncoder美元。在org. apache. sql应用（ExpressionEncoder. scala： 61）。编码器美元。产品（Encoders. scala： 274）在org. apache. spark. sql。SQLEncoders. newProductEncoder（SQLIncates. scala： 47）在sun.反射的人消费者$. main（人消费者. scala： 33）在sun.反射的人消费者. main（人消费者. scala）。NativemetodAccessorInm.调用0（本机方法）在sun.反射。N

以下是我的代码...

object PersonConsumer {
  import org.apache.spark.rdd.RDD
  import com.trueaccord.scalapb.spark._
  import org.apache.spark.sql.{SQLContext, SparkSession}
  import com.example.protos.demo._

  def main(args : Array[String]) {

    def parseLine(s: String): Person =
      Person.parseFrom(
        org.apache.commons.codec.binary.Base64.decodeBase64(s))

    val spark = SparkSession.builder.
      master("local")
      .appName("spark session example")
      .getOrCreate()

    import spark.implicits._

    val ds1 = spark.readStream.format("kafka").option("kafka.bootstrap.servers","localhost:9092").option("subscribe","person").load()

    val ds2 = ds1.selectExpr("CAST(value AS STRING)").as[String]

    val ds3 = ds2.map(str => parseLine(str)).createOrReplaceTempView("persons")

    val ds4 = spark.sqlContext.sql("select name from persons")

    val query = ds4.writeStream
      .outputMode("append")
      .format("console")
      .start()
    query.awaitTermination()
  }
}

Answer 1

在Person类中，性别是一个枚举，这就是导致此问题的原因。删除此字段后，它工作正常。以下是我从DataBricks的Shixiong（Ryan）那里得到的答案。

问题是“可选的性别性别=3；”。生成的类“性别”是一个特征，Spark不知道如何创建一个特征，因此不支持它。您可以定义SQL编码器支持的类，并将此生成的类转换为parseLine中的新类。

Answer 2

带val ds3的行应为：

val ds3 = ds2.map(str => parseLine(str))

sqlContext.protoToDataFrame(ds3).registerTempTable("persons")

在将RDD保存为临时表之前，需要将其转换为数据帧。