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

Question

我试图使用结构化流方法，使用基于DataFrame/DataSet API的Spark-Streaming来加载来自Kafka的数据流。

我使用：

• 火花2.10
• Kafka0.10
• SPARK-SQL-KAFKA-0-10

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

 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()

Answer 1

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

如果将数据序列化为JSON字符串，则有两个选项。您可以将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))

或强制转换到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: _*)