如何使用结构化流从 Kafka 读取 JSON 格式的记录？

Question

我正在尝试 structured streaming approach 使用基于 DataFrame/Dataset API 的 Spark-Streaming 从 Kafka 加载数据流。

我使用：

Spark 2.10
卡夫卡 0.10
spark-sql-kafka-0-10

Spark Kafka DataSource 已定义底层架构：

|key|value|topic|partition|offset|timestamp|timestampType|

我的数据采用 json 格式，它们存储在 value 列中。我正在寻找一种方法如何从值列中提取基础架构并将接收到的数据帧更新为存储在 value 中的列？我尝试了下面的方法，但它不起作用：

 val columns = Array("column1", "column2") // column names
 val rawKafkaDF = sparkSession.sqlContext.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers","localhost:9092")
  .option("subscribe",topic)
  .load()
  val columnsToSelect = columns.map( x => new Column("value." + x))
  val kafkaDF = rawKafkaDF.select(columnsToSelect:_*)

  // some analytics using stream dataframe kafkaDF

  val query = kafkaDF.writeStream.format("console").start()
  query.awaitTermination()

这里我得到异常 org.apache.spark.sql.AnalysisException: Can't extract value from value#337; 因为在创建流时，里面的值是未知的...

你有什么建议吗？

Answer 1

从 Spark 的角度来看 value 只是一个字节序列。它不知道序列化格式或内容。为了能够提取文件，您必须先解析它。

如果数据被序列化为 JSON 字符串，您有两个选择。您可以 cast value 到 StringType 并使用 from_json 并提供架构：

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions.from_json

val schema: StructType = StructType(Seq(
  StructField("column1", ???),
  StructField("column2", ???)
))

rawKafkaDF.select(from_json($"value".cast(StringType), schema))

或 cast 到 StringType，使用 get_json_object:

按路径提取字段

import org.apache.spark.sql.functions.get_json_object

val columns: Seq[String] = ???

val exprs = columns.map(c => get_json_object($"value", s"$$.$c"))

rawKafkaDF.select(exprs: _*)

和 cast 稍后到所需的类型。

如何使用结构化流从 Kafka 读取 JSON 格式的记录？

How to read records in JSON format from Kafka using Structured Streaming?

scala

apache-kafka

apache-spark

apache-spark-sql

spark-structured-streaming