时间戳

首页 > 解决方案 > 在 PySpark 中创建数据框时处理不同的 JSON 模式

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

问题描述

我有 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

解决方案

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)

所以这就是我所做的，

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

推荐阅读