Spark读取CSV文件时如何跳过前N行与后N行？及处理带首尾标识的API返回CSV数据的替代方案

算力阿峰

2026-4-28

问题1：Spark读取CSV时跳过前N行和后N行

跳过前N行

Spark提供了直接的配置项处理开头行的跳过，非常实用：

如果前N行是无关的注释/元数据，且第N+1行是CSV表头，可以这么写：

val df = spark.read
  .option("skipRows", N) // 跳过前N行
  .option("header", "true") // 将第N+1行作为表头
  .csv("path/to/your/file.csv")

如果前N行不包含表头，只是需要跳过无效行，直接设置skipRows即可，表头参数根据实际情况调整。

跳过最后N行

跳过末尾行相对复杂，因为Spark是分布式计算，无法直接获取总行数。这里有两种常用方案：

方案1：先获取总行数，再过滤行号

// 第一步：读取所有行，计算总行数
val totalRows = spark.read.text("path/to/your/file.csv").count()

// 第二步：给每行添加行号，过滤掉最后N行
import org.apache.spark.sql.functions.row_number
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.monotonically_increasing_id

val dfWithRowNum = spark.read
  .option("header", "true")
  .csv("path/to/your/file.csv")
  .withColumn("row_num", row_number().over(Window.orderBy(monotonically_increasing_id())))

val df = dfWithRowNum.filter(s"row_num <= ${totalRows - N}").drop("row_num")

注意：用monotonically_increasing_id()是因为如果没有天然排序键，它可以生成伪顺序（不保证连续，但不影响“去掉最后N行”的需求）。如果有可排序的业务字段，用该字段orderBy会更准确。

方案2：RDD zipWithIndex（适合小文件）
如果文件体积不大，可先转成RDD处理：

val rdd = spark.sparkContext.textFile("path/to/your/file.csv")
val totalRows = rdd.count()
val filteredRdd = rdd.zipWithIndex().filter(_._2 < totalRows - N).map(_._1)

// 将过滤后的RDD转回DataFrame
val df = spark.read
  .option("header", "true")
  .csv(filteredRdd.toDS())

这种方法在大文件场景下性能较差，因为需要在Driver端计算总行数，且zipWithIndex会触发shuffle。

问题2：读取带头部/主体/尾部标记的CSV数据

你的预处理方案是可行的，但Spark也支持直接在读取流程中过滤出BODY部分，无需提前修改原始文件。推荐方案是先以文本格式读取所有行，过滤出BEGIN_BODY与END_BODY之间的内容，再转成CSV DataFrame。

方案1：通过行号过滤

// 1. 读取所有文本行
val textDF = spark.read.text("path/to/your/data.csv")

// 2. 给每行添加行号，定位BEGIN_BODY和END_BODY的位置
import org.apache.spark.sql.functions.{col, row_number}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.monotonically_increasing_id

val dfWithRowNum = textDF.withColumn("row_num", row_number().over(Window.orderBy(monotonically_increasing_id())))

val startRow = dfWithRowNum.filter(col("value") === "----BEGIN_BODY").select("row_num").first().getLong(0)
val endRow = dfWithRowNum.filter(col("value") === "----END_BODY").select("row_num").first().getLong(0)

// 3. 过滤出BODY区间内的行，跳过标记行
val bodyRows = dfWithRowNum
  .filter(col("row_num") > startRow && col("row_num") < endRow)
  .select("value")

// 4. 将过滤后的行转成CSV DataFrame
val finalDF = spark.read
  .option("header", "true")
  .option("inferSchema", "true") // 可选：自动推断字段类型
  .csv(bodyRows.toDS())

finalDF.show()

方案2：RDD分区内状态过滤（性能更优）

利用RDD的mapPartitions在分区内标记是否进入BODY区间，无需计算行号，适合大文件：

val rdd = spark.sparkContext.textFile("path/to/your/data.csv")

val bodyRdd = rdd.mapPartitions(iter => {
  var inBody = false
  iter.filter { line =>
    if (line == "----BEGIN_BODY") {
      inBody = true
      false // 跳过标记行
    } else if (line == "----END_BODY") {
      inBody = false
      false // 跳过标记行
    } else {
      inBody
    }
  }
})

val finalDF = spark.read
  .option("header", "true")
  .csv(bodyRdd.toDS())

这种方法是分布式处理，不会触发额外shuffle，性能更优，尤其适合流水线式的数据处理流程。

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