Spark API如何读取含引号及单元格内分号的分号分隔CSV？

直接使用Spark CSV Reader的默认参数无法同时满足你的两个需求——默认解析会将未闭合引号内的所有内容视为单个单元格，而禁用引号解析（设置quote为特殊字符）又会拆分合法带引号单元格内的分号。以下两种方案可解决该问题：

方案一：自定义行解析逻辑 #

先以文本格式读取所有行，编写自定义函数识别成对引号，保留其内部的分号，同时将未闭合引号内的分号视为分隔符。

示例Scala代码：

import org.apache.spark.sql.functions._

// 读取原始文本行
val rawLines = spark.read.text("/sample.csv").select(col("value").alias("line"))

// 自定义CSV行拆分函数
def splitCsvLine(line: String, delimiter: Char = ';'): Array[String] = {
  val result = scala.collection.mutable.ListBuffer[String]()
  val current = new StringBuilder()
  var inQuotes = false
  
  for (c <- line) {
    c match {
      case '"' =>
        inQuotes = !inQuotes
        // 如需保留引号内容，可取消下一行注释
        // current.append(c)
      case `delimiter` if !inQuotes =>
        result += current.toString().trim
        current.clear()
      case _ =>
        current.append(c)
    }
  }
  // 添加最后一个字段
  result += current.toString().trim
  result.toArray
}

// 注册UDF用于DataFrame处理
val splitCsvUdf = udf(splitCsvLine _)

// 拆分字段并转换为多列DataFrame
val parsedDf = rawLines.withColumn("split_values", splitCsvUdf(col("line")))
  .select((0 until 3).map(i => col("split_values")(i).alias(s"col${i+1}")): _*)

parsedDf.show()

该函数的核心逻辑：

• 遇到成对双引号时，切换引号模式；
• 引号模式内的分号会被保留在单元格中，不做拆分；
• 非引号模式下的分号作为分隔符；
• 未闭合的引号会被当作普通字符处理，其内部的分号正常拆分。

方案二：预处理修复未闭合引号 #

如果数据中未闭合引号的情况较少，可先预处理修复不符合CSV规范的行，再用标准CSV解析逻辑处理：

示例Scala代码：

// 读取原始文本行
val rawLines = spark.read.text("/sample.csv").select(col("value").alias("line"))

// 修复未闭合引号：统计每行引号数量，奇数则移除行尾的引号
val fixedLines = rawLines.withColumn("quote_count", regexp_count(col("line"), "\""))
  .withColumn("fixed_line", when(col("quote_count") % 2 === 1, regexp_replace(col("line"), "\"$", ""))
    .otherwise(col("line")))
  .select("fixed_line")

// 将修复后的行写入临时路径
fixedLines.write.mode("overwrite").text("/tmp/fixed_sample.csv")

// 用标准CSV解析读取修复后的数据
val finalDf = spark.read
  .format("csv")
  .option("delimiter", ";")
  .option("quote", "\"")
  .option("escape", "\"")
  .option("ignoreLeadingWhiteSpace", true)
  .load("/tmp/fixed_sample.csv")

finalDf.show()

这种方式先修复非法的未闭合引号，再利用Spark原生CSV解析能力，既能保留合法带引号单元格内的分号，又能正常拆分其他分号。

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