找不到存储在数据集中的类型的编码器。在火花结构化流中

Question

我正在尝试 spark 网站上给出的 spark 结构化流的示例，但它抛出错误

1.找不到存储在数据集中的类型的编码器。通过导入 spark.implicits 支持原始类型（Int、String 等）和产品类型（案例类）。未来版本中将添加对序列化其他类型的支持。

2。方法的参数不足：（隐式证据$2：org.apache.spark.sql.Encoder[data]）org.apache.spark.sql.Dataset[data]。 未指定值参数evidence$2。 val ds: 数据集[数据] = df.as[数据]

这是我的代码

import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types._
import org.apache.spark.sql.Encoders
object final_stream {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession
          .builder()
          .appName("kafka-consumer")
          .master("local[*]")
          .getOrCreate()
        import spark.implicits._

        spark.sparkContext.setLogLevel("WARN")

    case class data(name: String, id: String)

    val df = spark
          .readStream
          .format("kafka")
          .option("kafka.bootstrap.servers", "172.21.0.187:9093")
          .option("subscribe", "test")
          .load()
    println(df.isStreaming)

    val ds: Dataset[data] = df.as[data]
    val value = ds.select("name").where("id > 10")




    value.writeStream
          .outputMode("append")
          .format("console")
          .start()
          .awaitTermination()

  }
}

任何关于如何使这项工作的帮助。？ 我想要这样的最终输出 我想要这样的输出

+-----+--------+
| name|    id
+-----+--------+
|Jacek|     1
+-----+--------+

Answer 1

错误的原因是您正在处理来自 Kafka 的 Array[Byte] 并且没有匹配 data 案例类的字段。

scala> println(schema.treeString)
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)

将df.as[data] 行更改为以下内容：

df.
  select($"value" cast "string").
  map(value => ...parse the value to get name and id here...).
  as[data]

我强烈建议使用select 和functions 对象来处理传入的数据。

Answer 2

该错误是由于数据框中的列数与您的案例类不匹配造成的。

您在数据框中有 [topic, timestamp, value, key, offset, timestampType, partition] 列

而您的案例类只有两列

case class data(name: String, id: String)

你可以将dataframe的内容显示为

val display = df.writeStream.format("console").start()

睡几秒钟，然后

display.stop()

并且还使用option("startingOffsets", "earliest") 提到的here

然后根据您的数据创建一个案例类。

希望这会有所帮助！