使用Apache Spark读取CSV时，如何定位导致行损坏的具体单元格？

在PERMISSIVE模式下，默认只会将整行标记为损坏并存入指定列，但要定位具体出错单元格，可采用以下几种方案：

方法一：逐列校验数据类型 #

针对每一列的预期数据类型，单独做校验，筛选出不符合规则的行和列：

• 基于开源Spark实现，遍历所有业务列（排除损坏记录列），对每一列尝试转换为预期类型，若原列非空但转换后为空，说明该单元格数据不符合要求，标记为损坏。

// Scala示例：标记转换失败的单元格
val df = spark.read
  .option("mode", "PERMISSIVE")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .csv("path/to/your.csv")

// 获取所有业务列
val businessColumns = df.columns.filter(_ != "_corrupt_record")

// 生成校验列，标记每个列的损坏状态
val validatedDf = businessColumns.foldLeft(df) { (acc, colName) =>
  acc.withColumn(s"${colName}_is_corrupt", 
    when(col(colName).cast("int").isNull && col(colName).isNotNull, true).otherwise(false)
  )
}

// 筛选存在损坏单元格的行
validatedDf.filter(businessColumns.map(c => s"${c}_is_corrupt").mkString(" OR ")).show()

• 根据实际需求调整cast的目标类型（如double、date等）。

方法二：对比原始行与解析后数据 #

利用损坏记录列的原始行数据，和解析后的各列数据逐字段对比，找出不匹配的单元格：

1. 将原始损坏行按CSV分隔符拆分（处理带引号的字段），得到原始字段数组。
2. 将解析后的列转为数组，逐元素对比两个数组，定位不一致或解析失败的位置。

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

val df = spark.read
  .option("mode", "PERMISSIVE")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .option("delimiter", ",") // 根据你的CSV分隔符调整
  .csv("path/to/your.csv")

val businessColumns = df.columns.filter(_ != "_corrupt_record")

// 拆分原始损坏行为字段数组（支持带引号的字段）
val dfWithRawFields = df.withColumn("raw_fields", split(col("_corrupt_record"), ",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)"))

// 转换解析后列为数组
val dfWithParsedFields = dfWithRawFields.withColumn("parsed_fields", array(businessColumns.map(col): _*))

// 对比数组，提取损坏单元格详情
val corruptCellsDf = dfWithParsedFields
  .select(
    col("_corrupt_record"),
    explode(
      array(
        businessColumns.zipWithIndex.map { case (colName, idx) =>
          when(
            col("raw_fields")(idx) =!= col("parsed_fields")(idx) || col("parsed_fields")(idx).isNull,
            concat(lit(s"列名: $colName, 原始值: "), col("raw_fields")(idx), lit(", 解析值: "), col("parsed_fields")(idx))
          ).otherwise(lit(null))
        }: _*
      )
    ).alias("corrupt_cell_detail")
  )
  .filter(col("corrupt_cell_detail").isNotNull)

corruptCellsDf.show(truncate = false)

• 若CSV存在复杂嵌套引号，可自定义UDF实现更精准的拆分逻辑。

方法三：自定义UDF做规则校验 #

编写自定义UDF，接收原始行和预期列结构，逐个字段校验自定义规则（如日期格式、数值范围），返回损坏单元格信息：

from pyspark.sql.functions import udf, lit
from pyspark.sql.types import StringType
import csv
from io import StringIO
from datetime import datetime

def find_corrupt_cells(raw_row, column_names, expected_types):
    reader = csv.reader(StringIO(raw_row))
    raw_fields = next(reader)
    corrupt_details = []
    for col_name, raw_val, exp_type in zip(column_names, raw_fields, expected_types):
        if not raw_val:
            continue
        try:
            if exp_type == "int":
                int(raw_val)
            elif exp_type == "float":
                float(raw_val)
            elif exp_type == "date":
                datetime.strptime(raw_val, "%Y-%m-%d")  # 替换为你的日期格式
        except ValueError:
            corrupt_details.append(f"列: {col_name}, 值: {raw_val} (不符合{exp_type}类型)")
    return "; ".join(corrupt_details) if corrupt_details else None

# 注册UDF
find_corrupt_udf = udf(find_corrupt_cells, StringType())

# 读取CSV并生成损坏单元格详情
df = spark.read \
    .option("mode", "PERMISSIVE") \
    .option("columnNameOfCorruptRecord", "_corrupt_record") \
    .csv("path/to/your.csv")

column_names = df.columns[:-1]  # 排除损坏记录列
expected_types = ["int", "string", "date", "float"]  # 对应各列的预期类型

df_with_corrupt_details = df.withColumn(
    "corrupt_cell_info",
    find_corrupt_udf(df["_corrupt_record"], lit(column_names), lit(expected_types))
)

df_with_corrupt_details.filter(df_with_corrupt_details["corrupt_cell_info"].isNotNull()).show(truncate=False)

以上方法均基于开源Spark实现，无需依赖Databricks平台，可精准定位导致行损坏的具体单元格。

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