Pyspark：从 JSON 文件创建架构

Question

我正在处理非常长的嵌套JSON文件中的数据。问题是，这些文件的结构并不总是相同的，因为有些文件缺少其他文件的列。我想从一个包含所有列的空JSON文件创建一个定制模式。如果我稍后将JSON文件读入这个预定义的模式，不存在的列将被空值填充(至少计划是这样的)。到目前为止我所做的:

1. 将测试 JSON（不包含预期的所有列）加载到数据帧中
2. 将其架构写入 JSON 文件
3. 在文本编辑器中打开此 JSON 文件并手动添加缺少的列

接下来我想做的是通过将JSON文件读入我的代码来创建一个新架构，但是我在合成器上遇到了困难。是否可以直接从文件本身读取架构？我试过

schemaFromJson = StructType.fromJson(json.loads('filepath/spark-schema.json'))

但它给了我TypeError： init（）缺少2个必需的位置参数：'doc'和'pos'

你知道我现在的代码有什么问题吗？非常感谢

编辑：我在 sparkbyexamples.com/pyspark/pyspark-structtype-and-structfield 遇到了这个链接。第7章几乎描述了我遇到的问题。我只是不明白如何解析我手动增强为模式的json文件从Json = 结构类型.从Json（json.loads（架构.json））。

当我这样做时：

jsonDF = spark.read.json(filesToLoad)
schema = jsonDF.schema.json()
schemaNew = StructType.fromJson(json.loads(schema))
jsonDF2 = spark.read.schema(schemaNew).json(filesToLoad)

代码可以运行，但显然没有用，因为jsonDF和jsonDF2有相同的内容/模式。我想要实现的是向“schema”添加一些列，然后这些列将反映在“schemaNew”中。

Answer 1

为什么不定义一个包含JSON文件可以包含的所有列的空DF呢？然后将JSONs加载到其中。这里有一个想法:

对于Spark 3.1.0：

from pyspark.sql.types import *

schema = StructType([
    StructField("fruit",StringType(),True),
    StructField("size",StringType(),True),
    StructField("color",StringType(),True)
])
df = spark.createDataFrame([], schema)

json_file_1 = {"fruit": "Apple","size": "Large"}
json_df_1 = spark.read.json(sc.parallelize([json_file_1]))

df = df.unionByName(json_df_1, allowMissingColumns=True)

json_file_2 = {"fruit": "Banana","size": "Small","color": "Yellow"}

df = df.unionByName(json_file_2, allowMissingColumns=True)

display(df)

Answer 2

我想我明白了。Schemapath包含已经增强的模式:

schemapath = '/path/spark-schema.json'
with open(schemapath) as f:
   d = json.load(f)
   schemaNew = StructType.fromJson(d)
   jsonDf2 = spark.read.schema(schmaNew).json(filesToLoad)
   jsonDF2.printSchema()