如何在Spark Scala中以类型安全方式读取多布局分段输入文件？

针对同一文件包含多种结构行的场景，我们可以通过定义强类型Case Class + 基于标识字段的模式匹配实现类型安全的解析，具体步骤如下：

1. 定义对应各分段的Case Class #

为每个分段的结构创建强类型Case Class，确保字段与分段定义完全匹配：

// 借款人（BS分段）
case class Borrower(indicator: String, firstName: String, lastName: String, companyName: String, joiningDate: String)
// 关联人（RS分段）
case class Relationship(indicator: String, name: String, role: String, companyName: String)
// 信贷额度（CR分段）
case class CreditFacility(indicator: String, loanType: String, accountNumber: String, sanctionedDate: String, amount: Double)

注：如果需要将日期字段转为java.sql.Date或LocalDate，可以在解析环节额外处理，此处先保留原始字符串格式。

2. 初始化SparkSession并读取原始数据 #

先将文件读取为字符串行的Dataset，由于每行结构不同，无法直接用CSV格式读取：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("MultiFormatDataParsing")
  .master("local[*]") // 本地测试用，生产环境移除该配置
  .getOrCreate()

import spark.implicits._

// 读取原始数据为字符串行的Dataset
val rawData = spark.read.textFile("path/to/your/data/file.txt")

3. 解析每行数据并映射到对应Case Class #

通过模式匹配，根据每行的第一个标识字段（BS/RS/CR），将拆分后的字段映射到对应Case Class，同时处理字段间的空格：

// 定义解析函数，返回Option处理无效行
def parseLine(line: String): Option[Any] = {
  // 按逗号+可选空格拆分字段，兼容CR行中带空格的字段
  val fields = line.split(",\\s*")
  fields.head match {
    case "BS" if fields.length == 5 =>
      Some(Borrower(fields(0), fields(1), fields(2), fields(3), fields(4)))
    case "RS" if fields.length == 4 =>
      Some(Relationship(fields(0), fields(1), fields(2), fields(3)))
    case "CR" if fields.length == 5 =>
      // 将金额字段转为Double类型
      Some(CreditFacility(fields(0), fields(1), fields(2), fields(3), fields(4).toDouble))
    case _ =>
      // 格式错误的行返回None，后续过滤
      None
  }
}

// 解析数据并过滤无效行
val parsedData = rawData.map(parseLine).filter(_.isDefined).map(_.get)

4. 拆分并转换为各类型的Dataset #

将混合类型的数据集拆分为三个独立的强类型Dataset，方便后续业务处理：

// 提取借款人数据集
val borrowers = parsedData.collect { case b: Borrower => b }.toDS()
// 提取关联人数据集
val relationships = parsedData.collect { case r: Relationship => r }.toDS()
// 提取信贷额度数据集
val creditFacilities = parsedData.collect { case c: CreditFacility => c }.toDS()

验证结果 #

通过show()方法查看各Dataset的解析结果：

borrowers.show()
relationships.show()
creditFacilities.show()

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