Spark Scala中展开DataFrame的CSV值列的技术方案问询

1. 移除文本限定符 #

假设最后一列的文本限定符是双引号（若为其他符号，只需调整正则表达式），用regexp_replace去掉首尾的限定符：

import org.apache.spark.sql.functions.{regexp_replace, split, col}

// 清洗目标列，移除首尾双引号
val cleanedDF = yourOriginalDF
  .withColumn("cleaned_csv", regexp_replace(col("your_last_col_name"), "^\"|\"$", ""))

注：如果限定符是单引号，正则改为^'|'$；若包含特殊字符（如\），需额外转义。

2. 将CSV字符串分割为数组 #

用split函数把清洗后的字符串按逗号分割成数组列：

val arrayDF = cleanedDF
  .withColumn("csv_array", split(col("cleaned_csv"), ","))

3. 动态生成展开列（CSV1~CSV9） #

因CSV长度不固定，预先定义最多展开到9列，不足位置自动填充null。通过循环生成列名和对应数组元素引用：

// 生成CSV1到CSV9的列定义
val expandedCols = (1 to 9).map(i => col("csv_array")(i - 1).alias(s"CSV$i"))

// 保留原DataFrame所有列，添加展开后的CSV列，清理临时列
val finalDF = arrayDF
  .select(yourOriginalDF.columns.map(col) ++ expandedCols: _*)
  .drop("cleaned_csv", "csv_array")

4. 导出无文本限定符的CSV #

直接使用Spark的CSV导出功能，此时输出文件中目标列已展开为多列，且无文本限定符：

finalDF.write
  .option("header", "true")
  .csv("your_output_path")

特殊场景处理 #

若CSV字符串内部包含转义逗号（如"a,b\"c,d"），直接分割会出错，可使用Spark内置的CSV解析器处理：

import org.apache.spark.sql.functions.from_csv
import org.apache.spark.sql.types.{StringType, StructType}

// 定义最多9个字符串列的CSV结构
val csvSchema = StructType((1 to 9).map(i => s"CSV$i" -> StringType))

// 自动解析带限定符的CSV，处理内部转义
val parsedDF = yourOriginalDF
  .withColumn("parsed_csv", from_csv(col("your_last_col_name"), csvSchema, Map("quote" -> "\"")))
  .select(yourOriginalDF.columns.map(col) ++ (1 to 9).map(i => col("parsed_csv")(s"CSV$i")): _*)

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