Apache Spark处理列名带固定模式的动态多列CSV坏记录方案咨询

方案1：轻量预取表头生成Schema（推荐） #

你之前顾虑的「重复加载两次文件」属于认知误区：预取表头不需要读取全量数据，Spark仅会拉取每个文件的第一行解析表头，开销极低，几乎不会产生额外性能损耗。
实现代码如下：

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.functions.col
import scala.util.matching.Regex

// 配置规则
val polColumnPattern: Regex = "^POL_\\d+$".r // POL列匹配规则
val fixedColumnDefine = Seq(
  "ID" -> StringType // 固定列的字段名和类型配置
)
val csvInputPath = "/tmp/your_csv_dir/*.csv"
val badRecordStoragePath = "/tmp/badRecordsPath"

// 仅提取表头，不加载全量数据
val headerOnlyDF = spark.read
  .option("header", "true")
  .csv(csvInputPath)

// 动态生成Schema
val dynamicPolFields = headerOnlyDF.columns
  .filter(colName => polColumnPattern.pattern.matcher(colName).matches())
  .map(polCol => StructField(polCol, IntegerType, nullable = true))
val finalSchema = StructType(
  fixedColumnDefine.map(f => StructField(f._1, f._2, nullable = true)) ++ dynamicPolFields
)

// 按生成的Schema读取全量数据，支持坏记录捕获
val finalDF = spark.read
  .option("header", "true")
  .option("badRecordsPath", badRecordStoragePath)
  .schema(finalSchema)
  .csv(csvInputPath)

该方案的优势：

• 完美兼容badRecordsPath的坏记录捕获能力，类型转换错误的记录会自动写入坏记录路径
• 无全量重复读取开销，仅多一次极轻量的表头拉取请求
• Schema完全动态生成，不需要硬编码任何列字段

方案2：全量读取后动态类型转换 #

如果文件数量极多，希望彻底避免两次文件请求，可以选择先统一按字符串读取所有列，再根据列名规则做类型转换：

import org.apache.spark.sql.functions.col
import scala.util.matching.Regex

val polColumnPattern: Regex = "^POL_\\d+$".r
val csvInputPath = "/tmp/your_csv_dir/*.csv"
val badRecordStoragePath = "/tmp/badRecordsPath"

// 全量读取，所有列默认按String类型解析
val rawDF = spark.read
  .option("header", "true")
  .option("inferSchema", "false")
  .option("badRecordsPath", badRecordStoragePath)
  .csv(csvInputPath)

// 动态生成列转换规则
val transformedColumns = rawDF.columns.map { colName =>
  if (polColumnPattern.pattern.matcher(colName).matches()) {
    col(colName).cast(IntegerType).alias(colName)
  } else {
    col(colName)
  }
}

val convertedDF = rawDF.select(transformedColumns:_*)

该方案注意事项：

• 类型转换失败的值会变为null，不会被badRecordsPath捕获，需要额外增加空值校验逻辑识别不合法记录
• 适合小批量文件读取场景

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