Spark Structured Streaming Databricks事件中心模式定义问题

Question

现在，我试图做相同的架构在Streread上。

val jsonSchema = StructType([ StructField("associatedEntities", struct<driver:StringType,truck:StringType>, True), 
                          StructField("heading", StringType, True), 
                          StructField("location", struct<accuracyType:StringType,captureDateTime:StringType,cityStateCode:StringType,description:StringType,latitude:DoubleType,longitude:DoubleType,quality:StringType,transmitDateTime:StringType>, True), 
                          StructField("measurements", array<struct<type:StringType,uom:StringType,value:StringType>>, True), 
                          StructField("source", struct<entityType:StringType,key:StringType,vendor:StringType>, True), 
                          StructField("speed", DoubleType, True)])

val df = spark
 .readStream
 .format("eventhubs")
 //.schema(jsonSchema) 
 .options(ehConf.toMap)
 .load()

当我在笔记本中运行此单元格时：15:错误：简单表达式val jsonSchema=StructType（[StructField（“associatedEntities”，struct，True），”的非法开始

编辑：目标是将数据放入数据框中。我可以从event hub消息的主体中获取json字符串，但如果无法使模式正常工作，我不确定该怎么做。

Answer 1

由于您的架构定义，您会收到错误消息。模式定义应该如下所示：

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

val jsonSchema = StructType(
                        Seq(StructField("associatedEntities", 
                                        StructType(Seq(
                                          StructField("driver", StringType), 
                                          StructField ("truck", StringType)
                                        ))),
                            StructField("heading", StringType),
                            StructField("measurements", ArrayType(StructType(Seq(StructField ("type", StringType), StructField ("uom", StringType), StructField("value", StringType)))))
                           )
                         )

您可以通过以下方式对模式进行双重检查：

jsonSchema.printTreeString

返回模式：

root
 |-- associatedEntities: struct (nullable = true)
 |    |-- driver: string (nullable = true)
 |    |-- truck: string (nullable = true)
 |-- heading: string (nullable = true)
 |-- measurements: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- type: string (nullable = true)
 |    |    |-- uom: string (nullable = true)
 |    |    |-- value: string (nullable = true)

正如在评论中提到的，你会得到二进制数据。因此，首先您将获得原始数据帧：

val rawData = spark.readStream
  .format("eventhubs")
  .option(...)
  .load()

你必须：

• 将数据转换为字符串
• 解析嵌套的json
• 把它压平

用解析的数据定义数据帧：

val parsedData = rawData
   .selectExpr("cast (Body as string) as json")
   .select(from_json($"json", jsonSchema).as("data"))
   .select("data.*")