我正在尝试使用structured streaming approach使用基于DataFrame/Dataset API的Spark-Streaming从Kafka加载数据流。
我使用:
- 星火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;
因为在创建流时,内部的值是未知的...
你有什么建议吗?
最佳答案
从 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: _*)
然后强制转换
到所需的类型。
关于scala - 如何使用结构化流从 Kafka 读取 JSON 格式的记录?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54835479/