如何使用Spark Scala及UDF将不规则分隔文本文件转为DataFrame

核心问题 #

输入文本使用**数量不固定的#**作为字段分隔符，无法直接用Spark默认CSV读取器处理，需要自定义分隔符匹配逻辑。

解决方案步骤 #

1. 读取原始文本文件
   先以纯文本格式读取整个文件，避开Spark默认CSV解析器的分隔符限制。
2. 分离表头与数据行
   提取第一行作为字段名，剩余行作为业务数据。
3. 自定义分隔符处理
   使用正则表达式#+匹配一个或多个连续的#，分割每行数据后过滤空字符串（避免连续#产生的无效空元素）。
4. 映射到指定Schema
   定义目标DataFrame的结构，将处理后的数据映射到对应字段。

完整代码示例 #

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType}

object MultiHashSeparatorProcessor {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .appName("MultiHashToDataFrame")
      .master("local[*]") // 本地调试用，生产环境移除
      .getOrCreate()
    
    import spark.implicits._

    // 1. 读取原始文本文件
    val textDS = spark.read.text("path/to/your/target/file.txt").as[String]

    // 2. 分离表头和数据行
    val header = textDS.first()
    val dataRows = textDS.filter(_ != header)

    // 3. 解析表头字段（处理多#分隔）
    val fields = header.split("#+").filter(_.nonEmpty)

    // 4. 定义目标Schema（可按需调整字段类型）
    val schema = StructType(
      fields.map(fieldName => {
        fieldName match {
          case "id" => StructField("id", IntegerType, nullable = false)
          case "salary" => StructField("salary", IntegerType, nullable = false)
          case _ => StructField(fieldName, StringType, nullable = true)
        }
      })
    )

    // 5. 处理数据行并转换为DataFrame
    val dataDF = dataRows.map(row => {
      // 分割每行数据，过滤空元素
      val values = row.split("#+").filter(_.nonEmpty)
      // 按字段位置转换类型（id、salary转Int，其余保留String）
      values.zipWithIndex.map { case (value, idx) =>
        idx match {
          case 0 | 2 => value.toInt
          case _ => value
        }
      }
    }).rdd.map(row => org.apache.spark.sql.Row.fromSeq(row))
      .toDF(schema)

    // 输出结果
    dataDF.show()

    spark.stop()
  }
}

关键细节说明 #

• 正则#+：精准匹配任意长度的连续#，解决分隔符数量不固定的问题。
• filter(_.nonEmpty)：剔除连续#分割后产生的空字符串，保证字段映射准确性。
• 类型转换：示例中根据字段逻辑将id和salary转为整数类型，其余字段保留字符串，可根据业务需求调整。

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