PySpark DataFrame Schema转换：将Struct内数组转为字符串

问题背景 #

通过以下代码生成了PySpark DataFrame df_array：

df_array = (df_csv.groupBy(df_csv.depth,df_csv.height,df_csv.weight,df_csv.width,df_csv.seller_id,df_csv.sku,df_csv.navigation_id,df_csv.category, 
                         df_csv.subcategory,df_csv.max_dimension_p,df_csv.max_side_p)
                         .agg(sf.struct(sf.collect_list('id_distribution_center').alias('id_distribution_center'),sf.collect_list('id_modality').alias('id_modality'),
                                    sf.collect_list('zipcode_initial').alias('zipcode_initial'),sf.collect_list('zipcode_final').alias('zipcode_final'),sf.collect_list('cost').alias('cost'),sf.collect_list('city').alias('city'),sf.collect_list('state').alias('state')).alias("infos_gerais_product")))

该DataFrame的原始schema如下：

root
 |-- depth: double (nullable = true)
 |-- height: double (nullable = true)
 |-- weight: double (nullable = true)
 |-- width: double (nullable = true)
 |-- seller_id: string (nullable = true)
 |-- sku: long (nullable = true)
 |-- navigation_id: long (nullable = true)
 |-- category: string (nullable = true)
 |-- subcategory: string (nullable = true)
 |-- max_dimension_p: double (nullable = true)
 |-- max_side_p: double (nullable = true)
 |-- infos_gerais_product: struct (nullable = false)
 |    |-- id_distribution_center: array (nullable = false)
 |    |    |-- element: long (containsNull = false)
 |    |-- id_modality: array (nullable = false)
 |    |    |-- element: long (containsNull = false)
 |    |-- zipcode_initial: array (nullable = false)
 |    |    |-- element: long (containsNull = false)
 |    |-- zipcode_final: array (nullable = false)
 |    |    |-- element: long (containsNull = false)
 |    |-- cost: array (nullable = false)
 |    |    |-- element: double (containsNull = false)
 |    |-- city: array (nullable = false)
 |    |    |-- element: string (containsNull = false)
 |    |-- state: array (nullable = false)
 |    |    |-- element: string (containsNull = false)

需要将infos_gerais_product结构体中的数组类型转换为字符串类型，目标schema如下：

|-- depth: double (nullable = true)
 |-- height: double (nullable = true)
 |-- weight: double (nullable = true)
 |-- width: double (nullable = true)
 |-- seller_id: string (nullable = true)
 |-- sku: long (nullable = true)
 |-- navigation_id: long (nullable = true)
 |-- category: string (nullable = true)
 |-- subcategory: string (nullable = true)
 |-- max_dimension_p: double (nullable = true)
 |-- max_side_p: double (nullable = true)
 |-- infos_gerais_product: struct (nullable = false)
 |    |-- id_distribution_center: string (nullable = false)
 |    |-- id_modality: string (nullable = false)
 |    |-- zipcode_initial: string (nullable = false)
 |    |-- zipcode_final: string (nullable = false)
 |    |-- cost: string (nullable = false)
 |    |-- city: string (nullable = false)
 |    |-- state: string (nullable = false)

尝试通过重新定义schema并使用spark.createDataFrame转换时，出现报错：

TypeError: data is already a DataFrame

错误原因 #

spark.createDataFrame()的第一个参数需要是RDD、列表等原始数据源，而非已存在的DataFrame。直接传入已有的DataFrame会触发类型错误，因为该方法不支持直接基于现有DataFrame重新指定schema。

解决方案 #

有两种可行的方式实现类型转换：

方式1：修改聚合逻辑，直接生成字符串类型字段 #

在agg阶段，将collect_list生成的数组直接通过array_join转换为字符串（用指定分隔符拼接数组元素），这样生成的结构体字段默认就是字符串类型：

df_array = (df_csv.groupBy(
    df_csv.depth, df_csv.height, df_csv.weight, df_csv.width,
    df_csv.seller_id, df_csv.sku, df_csv.navigation_id,
    df_csv.category, df_csv.subcategory, df_csv.max_dimension_p, df_csv.max_side_p
).agg(
    sf.struct(
        sf.array_join(sf.collect_list('id_distribution_center'), ",").alias('id_distribution_center'),
        sf.array_join(sf.collect_list('id_modality'), ",").alias('id_modality'),
        sf.array_join(sf.collect_list('zipcode_initial'), ",").alias('zipcode_initial'),
        sf.array_join(sf.collect_list('zipcode_final'), ",").alias('zipcode_final'),
        sf.array_join(sf.collect_list('cost'), ",").alias('cost'),
        sf.array_join(sf.collect_list('city'), ",").alias('city'),
        sf.array_join(sf.collect_list('state'), ",").alias('state')
    ).alias("infos_gerais_product")
))

# 验证schema
df_array.printSchema()

方式2：对现有DataFrame进行字段转换 #

如果已经生成了df_array，可以通过withColumn重新构造infos_gerais_product结构体，将每个数组字段转换为字符串：

from pyspark.sql import functions as sf

df2 = df_array.withColumn(
    "infos_gerais_product",
    sf.struct(
        sf.array_join(df_array.infos_gerais_product.id_distribution_center, ",").alias('id_distribution_center'),
        sf.array_join(df_array.infos_gerais_product.id_modality, ",").alias('id_modality'),
        sf.array_join(df_array.infos_gerais_product.zipcode_initial, ",").alias('zipcode_initial'),
        sf.array_join(df_array.infos_gerais_product.zipcode_final, ",").alias('zipcode_final'),
        sf.array_join(df_array.infos_gerais_product.cost, ",").alias('cost'),
        sf.array_join(df_array.infos_gerais_product.city, ",").alias('city'),
        sf.array_join(df_array.infos_gerais_product.state, ",").alias('state')
    )
)

# 验证schema
df2.printSchema()

注：如果需要其他分隔符，可以替换array_join的第二个参数（比如";"）；如果需要保留数组的原始格式（如带方括号），可以使用sf.to_json()代替array_join。

内容的提问来源于stack exchange，提问作者Vivian