从 json 列嵌套对象中删除 Pyspark Dataframe 中的重复列名

Question

所以我有一个数据框，它是从我数据库中的 table 收集的。当我从数据库中读取它时，json 列在我的数据框中变成了一个字符串，没问题我使用以下方法转换它：

  df_json = spark.read.json(df.rdd.map(lambda r: r.JsonCol))
  df = df.withColumn("json_data", from_json("JsonCol", df_json.schema)).drop("JsonCol")

现在我得到了我的数据框的以下架构，我已经从 json 中删除了很多不相关的属性，但是相信我有很多：

 root
 |-- Id: integer (nullable = true)
 |-- json_data: struct (nullable = false)
 |    |-- object: struct (nullable = true)
 |    |    |-- Id: long (nullable = true)
 |    |    |-- Id: string (nullable = true)
 |    |    |-- ... many more properties

这里有问题，在嵌套对象 object 中，我有两个同名的属性。这是因为数据库中一些较旧的条目将 object 的 Id 作为字符串类型，但后来更改为 long 类型.因此，我无法使用以下方式编写我的数据框：

df.write.format("delta").option('mergeSchema', 'true').option("inferSchema", "true").option("overwriteSchema", "true").mode("overwrite").save(output_folder)

我收到以下错误：

AnalysisException: Found duplicate column(s) in the metadata update: json.object.id;

我该如何解决这个问题。理想情况下，我想合并这两列名称，方法是将 Id 与字符串类型转换为 long 类型并合并它们。所以我可以获得如下模式：

 root
 |-- Id: integer (nullable = true)
 |-- json_data: struct (nullable = false)
 |    |-- object: struct (nullable = true)
 |    |    |-- Id: long (nullable = true)
 |    |    |-- ... many more properties

请注意 json 的根还有一个名为 Id 的属性，我希望它保持不变！

有人可以帮我解决这个问题吗？

Answer 1

而不是在使用后修改并删除具有相同名称的重复列：

df = df.withColumn("json_data", from_json("JsonCol", df_json.schema)).drop("JsonCol")

我采用了一种解决方案，我事先在 JsonCol 上使用了正则表达式替换：

df = df.withColumn("JsonCol", regexp_replace(col("JsonCol"), r'"Id":"[0-9]+"', '"Id":[0-9]+'))

这会删除 ID 两边的 "，这意味着它们将被解释为 long 而不是字符串。

综合起来：

df = df.withColumn("JsonCol", regexp_replace(col("JsonCol"), r'"Id":"[0-9]+"', '"Id":[0-9]+'))
df_json = spark.read.json(df.rdd.map(lambda r: r.JsonCol))
df = df.withColumn("json_data", from_json("JsonCol", df_json.schema)).drop("JsonCol")

这给了我预期的输出，之后我能够正确保存它：

root
 |-- Id: integer (nullable = true)
 |-- json_data: struct (nullable = false)
 |    |-- object: struct (nullable = true)
 |    |    |-- Id: long (nullable = true)

从 json 列嵌套对象中删除 Pyspark Dataframe 中的重复列名

Remove duplicate column name in a Pyspark Dataframe from a json column nested object

apache-spark-sql

pyspark

pyspark-dataframes