这里有一种方法可以遍历这些字段并动态修改它们的名称。首先使用main.schema.fields[0].dataType.fields访问目标字段。接下来，使用pythonmap将pre_添加到每个字段：

from pyspark.sql.types import *
from pyspark.sql.functions import col

inner_fields = main.schema.fields[0].dataType.fields

# [StructField(a,LongType,true),

# StructField(b,LongType,true),

# StructField(c,LongType,true),

# StructField(e,StringType,true),

# StructField(f,StringType,true),

# StructField(g,StringType,true)]

pre_cols = list(map(lambda sf: StructField(f"pre_{sf.name}", sf.dataType, sf.nullable), inner_fields))

new_schema = StructType(pre_cols)

main.select(col("MAIN_COL").cast(new_schema)).printSchema()

# root

# |-- MAIN_COL: struct (nullable = false)

# |    |-- pre_a: long (nullable = true)

# |    |-- pre_b: long (nullable = true)

# |    |-- pre_c: long (nullable = true)

# |    |-- pre_e: string (nullable = true)

# |    |-- pre_f: string (nullable = true)

# |    |-- pre_g: string (nullable = true)

最后，您可以将cast与@Mahesh已经提到的新模式一起使用。

Spark的美丽，您可以通过编程操作元数据

这是一个继续原始代码片段的示例：

main.createOrReplaceTempView("table_name")

new_cols_select = ", ".join(["MAIN_COL." + col + " as pre_" + col for col in spark.sql("select MAIN_COL.* from table_name").columns])

new_df = spark.sql(f"select {new_cols_select} from table_name")

由于Spark的懒惰，并且因为所有操作都只是元数据，所以该代码几乎没有任何性能成本，并且对于10列或500列也同样有效(我们实际上正在对1k的列执行类似的操作)。

还可以使用df.schema对象以更优雅方式获取原始列名

您可以尝试这样做：根据需要将所有列添加到方案2

val schema2 = new StructType()
    .add("pre_a",StringType)
    .add("pre_b",StringType)
    .add("pre_c",StringType)

现在使用LIKE选择列：

df.select(col("MAIN_COL").cast(schema2)).show()

它将为您提供所有更新的列名。

下面的展开所有结构列添加父列名称作为前缀。

struct_cols = [c for c, t in df.dtypes if t.startswith('struct')]
for c in struct_cols:
    schema = T.StructType([T.StructField(f"{c}_{f.name}", f.dataType, f.nullable) for f in df.schema[c].dataType.fields])
    df = df.withColumn(c, F.col(c).cast(schema))
df = df.select([f"{c}.*" if c in struct_cols else c for c in df.columns])

测试输入：

from pyspark.sql import functions as F
from pyspark.sql import types as T

df = spark.createDataFrame([((1, 2), 5)], 'c1:struct<f1:int,f2:int>, c2:int')
print(df.dtypes)

# [('c1', 'struct<f1:int,f2:int>'), ('c2', 'int')]

结果：

struct_cols = [c for c, t in df.dtypes if t.startswith('struct')]
for c in struct_cols:
    schema = T.StructType([T.StructField(f"{c}_{f.name}", f.dataType, f.nullable) for f in df.schema[c].dataType.fields])
    df = df.withColumn(c, F.col(c).cast(schema))
df = df.select([f"{c}.*" if c in struct_cols else c for c in df.columns])

print(df.dtypes)

# [('c1_f1', 'int'), ('c1_f2', 'int'), ('c2', 'int')]

您也可以使用PySpark执行此操作：

df.select([col(col_name).alias('prefix' + col_name) for col_name in df])