Spark：如何在嵌套结构字段中动态重命名特殊字符

Question

我阅读了大量嵌套很深的带有字段的json，其中包含特殊字符，这会造成很多麻烦。

我想在 PySpark 中将字段的字符 / 和 - 重命名为下划线 _。例如列a-new 到a_new··

注意：有数千个带有特殊字符的字段名称，因此应该动态完成。如果只在反引号中添加字段更容易解决问题，这也是解决方案。我面临的问题是 spark 仅解释结构名称的一部分（a-new 为 a 等）。

参考：Rename nested field in spark dataframe

输入df：

root
 |-- a-new: long (nullable = true)
 |-- b/old: struct (nullable = true)
 |    |-- c-red: struct (nullable = true)
 |    |    |-- d/bue: struct (nullable = true)
 |    |    |    |-- e-green: string (nullable = true)
 |    |    |    |-- f-white: struct (nullable = true)
 |    |    |    |    |-- g/blue: array (nullable = true)
 |    |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |    |-- date: long (nullable = true)
 |    |    |    |    |    |    |-- val: long (nullable = true)

要求的结果：

root
 |-- a_new: long (nullable = true)
 |-- b_old: struct (nullable = true)
 |    |-- c_red: struct (nullable = true)
 |    |    |-- d_bue: struct (nullable = true)
 |    |    |    |-- e_green: string (nullable = true)
 |    |    |    |-- f_white: struct (nullable = true)
 |    |    |    |    |-- g_blue: array (nullable = true)
 |    |    |    |    |    |-- element: struct (containsNull = true)
 |    |    |    |    |    |    |-- date: long (nullable = true)
 |    |    |    |    |    |    |-- val: long (nullable = true)

我想知道是否有比我在解决方案中找到的使用新模式重新创建 df 更有效的方法： https://stackoverflow.com/a/58030523/9579821

json_1 = """{"a-new":1,"b/old":{"c-red":{"d/bue":{"e-green":"label_1","f-white":{"g/blue":[{"date":2020,"val":1}]}}}}}"""
df = spark.read.json(sc.parallelize([json_1]))
df.printSchema()

    # Some imports
from pyspark.sql.types import DataType, StructType, ArrayType
from copy import copy

# We take a dataframe and return a new one with required changes
def clean_df(df):
    # Returns a new sanitized field name (this function can be anything really)
    def sanitizeFieldName(s: str) -> str:
        return s.replace("-", "_").replace("/", "_")
    
    # We call this on all fields to create a copy and to perform any 
    # changes we might want to do to the field.
    def sanitizeField(field: StructField) -> StructField:
        field = copy(field)
        field.name = sanitizeFieldName(field.name)
        # We recursively call cleanSchema on all types
        field.dataType = cleanSchema(field.dataType)
        return field
    
    def cleanSchema(dataType: [DataType]) -> [DataType]:
        dataType = copy(dataType)
        # If the type is a StructType we need to recurse otherwise 
        # we can return since we've reached the leaf node
        if isinstance(dataType, StructType):
            # We call our sanitizer for all top level fields
            dataType.fields = [sanitizeField(f) for f in dataType.fields]
        elif isinstance(dataType, ArrayType):
            dataType.elementType = cleanSchema(dataType.elementType)
        return dataType

    # Now since we have the new schema we can create a new DataFrame 
    # by using the old Frame's RDD as data and the new schema as the 
    # schema for the data
   
  return spark.createDataFrame(df.rdd, cleanSchema(df.schema))
   
clean_df(df).printSchema()

Answer 1

   # Rename columns using `withColumnRenamed`
   for c in df.columns:

        df = df.withColumnRenamed(c,c.replace('-','_').replace('/','_'))
     
   # Rename nested fields using `cast`
   for c in df.columns:
        new_schema = df.select(c).schema.simpleString().replace('-','_').replace('/','_')[8+len(c):-1]
        df = df.withColumn(c,F.col(c).cast(new_schema))