PySpark列转换导致数据错乱问题求助

需求说明 #

需要将PySpark DataFrame中的数组列（如[1,2,3]）转换为"(1;2;3)"格式的字符串，同时对所有列执行正则处理：移除换行符、回车符、~符号；空数组转换后生成的()也需要移除；非数组列仅执行正则移除操作。

示例输入 #

期望输出 #

问题现象 #

遍历列进行处理后，出现数据错乱，部分行的数组列内容被替换为其他行的数据。

原尝试代码 #

for c in df.columns:
    if isinstance(df.schema[c].dataType, ArrayType):
        print(c)
        df= df.withColumn(c, concat_ws(';', col(c))).withColumn(c, concat(lit("("), col(c), lit(")"))).withColumn(c, F.regexp_replace(c, '\n|\r|\\n|\\r|~|\\(\\)|', ''))
    else:
        df= df.withColumn(c, F.regexp_replace(c, '\n|\r|\\n|\\r|~|', ''))

解决方案 #

问题根源 #

1. 正则表达式错误：末尾多余的|会匹配空字符串，导致对原有字符串进行无意义的重复替换，破坏数据结构。
2. 低效的列处理方式：循环中多次调用withColumn生成新DataFrame，可能导致执行计划复杂，增加出错概率。

修正后的代码 #

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

# 构造所有列的处理表达式
processed_columns = []
for col_name in df.columns:
    col_type = df.schema[col_name].dataType
    
    if isinstance(col_type, T.ArrayType):
        # 数组列处理：拼接元素→加括号→正则清理
        processed_col = F.regexp_replace(
            F.concat(F.lit("("), F.concat_ws(";", F.col(col_name)), F.lit(")")),
            r'[\n\r~]|^\(\)$',  # 匹配换行、回车、~，或空括号（空数组场景）
            ''
        )
    else:
        # 非数组列仅清理特殊字符
        processed_col = F.regexp_replace(F.col(col_name), r'[\n\r~]', '')
    
    processed_columns.append(processed_col.alias(col_name))

# 一次性生成处理后的DataFrame
df_processed = df.select(*processed_columns)

代码说明 #

• 正则优化：用字符类[\n\r~]简化重复的换行/回车匹配，用^\(\)$精准匹配空括号（空数组转换后的结果），避免误替换。
• 批量列处理：先构建所有列的处理逻辑，再通过select一次性生成结果，减少执行计划的复杂度，避免数据错乱。

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