如何用Spark Scala加载无分隔符文本文件并动态按列长转存CSV？

核心思路 #

通过动态定义字段与对应长度的映射关系，结合Spark的字符串截取函数，将每行无分隔符文本拆分为指定字段，最终保存为带分隔符的CSV文件。

实现步骤与代码示例 #

1. 定义动态列长度配置

先创建包含字段名和对应长度的列表，这个配置可根据需求灵活修改：

// 动态列配置：(字段名, 长度)
val columnConfigs = List(("Name", 50), ("address", 40), ("age", 2))

2. 加载无分隔符文本文件

使用Spark的text数据源加载原始文本，每行数据会被封装在名为value的列中：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("FixedLengthToCSV")
  .master("local[*]") // 生产环境移除该行
  .getOrCreate()

// 加载无分隔符文本文件
val rawDF = spark.read.text("path/to/your/fixed-length-file.txt")

3. 动态拆分字段

通过foldLeft迭代列配置，逐步为DataFrame添加截取后的字段，同时维护起始索引，每次截取后更新索引位置：

import org.apache.spark.sql.functions.{substring, col}

// 初始起始索引为1（Spark的substring从1开始计数）
val processedDF = columnConfigs.foldLeft((rawDF, 1)) { case ((df, startIdx), (colName, length)) =>
  val newDF = df.withColumn(colName, substring(col("value"), startIdx, length))
  (newDF, startIdx + length)
}._1

// 移除原始的value列
val finalDF = processedDF.drop("value")

4. 保存为CSV文件

将处理后的DataFrame保存为CSV，设置表头和分隔符：

finalDF.write
  .option("header", "true")
  .option("delimiter", ",")
  .mode("overwrite") // 根据需求选择模式：append/overwrite/ignore等
  .csv("path/to/save/output.csv")

关键细节说明 #

• 动态适配：通过columnConfigs列表可任意新增、修改字段和长度，无需修改核心拆分逻辑
• 字符串索引：Spark的substring函数起始索引从1开始，而非0，需注意避免截取错误
• 边界处理：若原始行长度不足配置的总长度，substring会返回现有字符（不会报错），如果需要处理这种情况，可添加判断逻辑：

  // 可选：当行长度不足时填充null
  import org.apache.spark.sql.functions.{length, when, lit}
  val safeProcessedDF = rawDF.withColumn("row_length", length(col("value")))
    .transform(df => columnConfigs.foldLeft((df, 1)) { case ((currentDF, startIdx), (colName, length)) =>
      val endIdx = startIdx + length - 1
      val newCol = when(col("row_length") >= endIdx, substring(col("value"), startIdx, length))
        .otherwise(lit(null))
      (currentDF.withColumn(colName, newCol), startIdx + length)
    }._1)
    .drop("value", "row_length")
  

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