在 PySpark 中创建数据框时处理不同的 JSON 模式答案

【问题标题】：Handling varying JSON schema when creating a dataframe in PySpark在 PySpark 中创建数据框时处理不同的 JSON 模式
【发布时间】：2021-05-17 04:47:24
【问题描述】：

我有一个 Databricks 笔记本，它每小时读取一次 JSON 格式的增量数据。所以让我们在上午 11 点说文件的架构如下，

root
 |-- number: string (nullable = true)
 |-- company: string (nullable = true)
 |-- assignment: struct (nullable = true)
 |    |-- link: string (nullable = true)
 |    |-- value: string (nullable = true)

下一个小时的中午 12 点，架构更改为，

root
 |-- number: string (nullable = true)
 |-- company: struct (nullable = true)
 |    |-- link: string (nullable = true)
 |    |-- value: string (nullable = true)
 |-- assignment: struct (nullable = true)
 |    |-- link: string (nullable = true)
 |    |-- value: string (nullable = true)

某些列从字符串更改为结构，反之亦然。因此，如果我选择 col(company.link) 并且传入的模式是字符串类型，则代码将失败。在读取文件时如何处理 PySpark 中的架构更改，因为我的最终目标是将 JSON 展平为 CSV 格式。

【问题讨论】：

标签： pyspark

【解决方案1】：

def get_dtype(df,colname):
  return [dtype for name, dtype in df.dtypes if name == colname][0]

#df has the exploded JSON data
df2 = df.select("result.number",
                "result.company",             
                "result.assignment_group")
df23 = df2

for name, cols in df2.dtypes:
  if 'struct' in get_dtype(df2, name):
    try:
      df23 = df23.withColumn(name+"_link", col(name+".link")).withColumn(name+"_value", col(name+".value")).drop(name)
    except:
      print("error")

df23.printSchema()
root
 |-- number: string (nullable = true)
 |-- company: string (nullable = true)
 |-- assignment_group_link: string (nullable = true)
 |-- assignment_group_value: string (nullable = true)

这就是我所做的，

创建了一个函数，用于识别列是否为 struct 类型
从具有 JSON 分解结果的基本数据帧中读取所有列
然后循环遍历列，如果它是 struct 类型，则添加具有嵌套值的新列。

【讨论】：