TypeError：StructField 类型的对象不是 JSON 可序列化的答案

【问题标题】：TypeError: Object of type StructField is not JSON serializableTypeError：StructField 类型的对象不是 JSON 可序列化的
【发布时间】：2021-09-27 02:24:18
【问题描述】：

我正在尝试使用来自 Azure 事件中心的 json 数据流，以便通过 Databricks 上的 PySpark 进行进一步处理以进行分析。

我在尝试将 json 数据提取到笔记本中的数据帧时遇到问题。

我可以成功连接到事件中心，并且可以看到数据流到达。

为了解释源 json 结构，请参阅下面发送到事件中心的 json 数据版本。

 [
    {
       "header": {
          "msg_type": "0003",
          "source_dev_id": "TP1",
          "user_id": "RP1156"
       },
       "body": {
          "event_type": "ARRIVAL",
          "gbtt_timestamp": "1626464040000",
          "original_loc_stanox": "FELIX0008"
       }
    },
    {
       "header": {
          "msg_type": "0003",
          "source_dev_id": "TP1",
          "user_id": "RP1156"
       },
       "body": {
          "event_type": "ARRIVAL",
          "gbtt_timestamp": "1626465080000",
          "original_loc_stanox": "CREW0008"
       }
    },
    {
       "header": {
          "msg_type": "0002",
          "source_dev_id": "",
          "user_id": "RP1156"
       },
       "body": {
          "event_type": "DEPATURE",
          "gbtt_timestamp": "1626466070000",
          "original_loc_stanox": "FELIX0008"
       }
    }
 ]

以下是我连接到偶数集线器后在笔记本中做的事情：

# Read the stream to a data frames
df = spark.readStream.format("eventhubs").options(**ehConf).load()

# display(df) shows me that the data is arrving

# the structure of df at this stage is as follows
df.printSchema()
root
 |-- body: binary (nullable = true)
 |-- partition: string (nullable = true)
 |-- offset: string (nullable = true)
 |-- sequenceNumber: long (nullable = true)
 |-- enqueuedTime: timestamp (nullable = true)
 |-- publisher: string (nullable = true)
 |-- partitionKey: string (nullable = true)
 |-- properties: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)
 |-- systemProperties: map (nullable = true)
 |    |-- key: string
 |    |-- value: string (valueContainsNull = true)

# all the json data is in body as binary which I cast to string later on

# next I try to extract the json data into a data frame

from pyspark.sql.types import *
import  pyspark.sql.functions as F

dfsch = ArrayType(
    StructField("body",StructType([
        StructField("event_type",StringType(),True),
        StructField("gbtt_timestamp",StringType(),True),
        StructField("original_loc_stanox",StringType(),True)
    ]),True),
    StructField("header",StructType([
        StructField("msg_type",StringType(),True),
        StructField("source_dev_id",StringType(),True),
        StructField("user_id",StringType(),True)
    ]),True)
)

dfj = df.select(F.from_json(F.col("body").cast("string"), dfsch).alias("payload"))

# at this stage I get the following error

TypeError: Object of type StructField is not JSON serializable

我收到以下错误：

TypeError：StructField 类型的对象不是 JSON 可序列化的

我对 pyspark 很陌生，所以不确定我是否在 df3sch 中正确表示了 json 架构？

如果有人能指出这里有什么问题，将不胜感激？

【问题讨论】：

df.select(F.col("body").cast("string")).show() 输出什么？
我不能做 .show() 因为它是一个流数据帧，所以我做了以下事情： dfs = df.select(F.col("body").cast("string") ) display(dfs) 我当然可以在 body 列中看到与我之前给出的原始来源相同的 json 字符串。但我不确定如何将 json 字符串转换为数据框，以便进一步处理实际的列值？

标签： pyspark databricks spark-structured-streaming azure-eventhub

【解决方案1】：

更仔细地查看架构定义，它实际上很简单。我只需要将架构更改为以下内容：

dfsch = ArrayType(
    StructType([
        StructField("body",StructType([
            StructField("event_type",StringType(),True),
            StructField("gbtt_timestamp",StringType(),True),
            StructField("original_loc_stanox",StringType(),True)
        ])),
        StructField("header",StructType([
            StructField("msg_type",StringType(),True),
            StructField("source_dev_id",StringType(),True),
            StructField("user_id",StringType(),True)
        ]))
    ])
)

表明我还是 pyspark 的新手。向上和向上！

感谢您的帮助。

【讨论】：