Pyspark:从 JSON 文件创建模式

Question

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

1. 将测试 JSON(不包含预期的所有列)加载到数据框中
2. 将其模式写入 JSON 文件
3. 在文本编辑器中打开此 JSON 文件并手动添加缺失的列

接下来我想做的是通过将 JSON 文件读入我的代码来创建一个新模式，但我在语法方面遇到了困难。我可以直接从文件本身读取架构吗？我试过了

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

但它给了我 TypeError: init() missing 2 required positional arguments: 'doc' and 'pos'

知道我当前的代码有什么问题吗？非常感谢

编辑:我遇到了这个链接 sparkbyexamples.com/pyspark/pyspark-structtype-and-structfield 。第 7 章几乎描述了我遇到的问题。我只是不明白如何解析我手动增强为 schemaFromJson = StructType.fromJson(json.loads(schema.json)) 的 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”添加一些列，这些列随后将反射(reflect)在“schemaNew”中。

Answer 1

我想我明白了。 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()