json 文件演变为三角洲湖的动态模式

Question

我正在建立一个Azure Databricks delta-lake，并且正在努力将我的json数据加载到delta-lake中。json中有100多种不同的文件格式。全部储存在数据湖中。

现在，我试图避免编写100个不同的python笔记本，而是构建一个元数据驱动的笔记本，它应该能够处理所有不同的json格式。

我能够得到进入三角洲湖的第一批数据，到目前为止一切顺利。问题是当我加载到特定delta-lake表中的第一个json文件在一列中包含NULL时。然后，对delta-lake的写入会自动将该列创建为string。下一个文件在同一列中包含一个嵌套的json数组，结果是我得到了这个错误消息:

html" target="_blank">分析异常：无法合并字段“payment_info”和“payment_info”。无法合并不兼容的数据类型StringType和structType（structField（@type， StringType， true）， structField（bank_name， StringType， true）， structField（bic， StringType， true）， structField（iban， StringType， true）， structField（所有者， StringType， true））

这是我的数据。首先json看起来像：

{
"payment_info": null,
"shop_id": 1,
"shop_name": "Testshop",
"shop_state": "OPEN"
}

然后第二个json文件包含以下信息：

{
"payment_info": {
    "@type": "IBAN",
    "bank_name": "bankName",
    "bic": "ABCD12345",
    "owner": "baName"
},
"shop_id": 2,
"shop_name": "Another TestShop",
"shop_state": "OPEN"

}

我猜这是代码的相关部分:

jsondf = spark.read.option("multiline","true") \
  .format("json") \
  .option("mergeSchema", "true") \
  .load(load_path)

jsondf.write \
  .format("delta") \
  .mode("append") \
  .option("mergeSchema", "true") \
  .saveAsTable(table_name)

如果我能只创建初始增量表，而不创建所有记录的 NULL 列，我会非常高兴。这样，架构演化将在稍后填充时起作用。我没有任何情况，列首先作为int，然后作为字符串，或者首先作为字符串，然后作为复杂结构。

我还试着把它放在一个临时表中:

jsondf.createOrReplaceTempView("tempview")
sourcedf = spark.sql("select * from tempview")

并改用合并语句：

spark.sql("set spark.databricks.delta.schema.autoMerge.enabled=true")
spark.sql("MERGE INTO " + table_name + " as t " + \
  "USING " + batch_table_name + " as s " + \
  "ON " + joinsql + " " + \
  "WHEN MATCHED THEN " + \
     "UPDATE SET * " + \
  "WHEN NOT MATCHED THEN " + \
     "INSERT *")

不幸的是，这一切都以相同的结果告终。

那么，有没有什么简单的方法可以删除所有为空的列？或者有别的方法吗？

也许我可以使用一个填充了所有字段的伪json文件来获得表的模式定义。那是可行的。而手动定义模式将非常耗时。

干杯

Answer 1

这是两个月前的事了，但我是这样处理这个问题的。

读取目录中的所有 json 文件。将架构保存到一个文件中，以便在读取目录和写入增量的单独作业中引用。

from pyspark.sql.types import StructType    
import json    

file_schema = spark.read.format("json").load("s3a://dir/*.json")

file_schema.printSchema()

with open("/dbfs/FileStore/schemas/schema.json", "w") as f:
    json.dump(file_schema.schema.jsonValue(), f)

现在在新作业中，我加载模式文件并在读取时使用它

with open("/dbfs/FileStore/schemas/schema.json") as f:
    the_schema = StructType.fromJson(json.load(f))

然后，您可以在模式选项中引用它

file_reader = spark.readStream.format('json') \
    .schema(gds_schema) \
    .load(your_path_to_files)

这是一个经过清理的版本，但是它为您提供了一个正确的方向，并且有一个可以引用的托管模式。如果您的文件具有相同的字段名但不同的值类型，您会希望它保存为字符串。

但是，我建议只使用自动加载程序作为用例，让自动加载程序将字段存储为字符串，并在下游应用您的转换。我每天将其用于250k新的json文件，其中包含超过350个字段的架构。