PySpark读取CSV时如何忽略列数超过3的记录？

源CSV文件内容：

c1,c2,c3
1,4,46
2,6,246
4,7,542,87
5,7,623
5,7,623,32432

期望输出：

c1,c2,c3
1,4,46
2,6,246
5,7,623

请问将文件读取为DataFrame时，如何移除/忽略列数超过3的记录？spark.read.csv API是否有特定选项？我尝试过maxColumns选项，但它直接抛出错误。

如果Spark读取API没有对应选项，恳请告知解决方案。我想到了如下方案：

spark.read.text().split(',').alias('columncount').filter(" columncount > 3")

但如果能在读取时使用内置函数/选项处理会更好。

Spark CSV读取API的限制 #

Spark 的 spark.read.csv 并没有内置选项可以直接忽略列数不匹配的记录。你使用的 maxColumns 参数是用来限制CSV文件允许的最大列数，当文件中存在列数超过该值的行时，Spark 会直接抛出异常，而非过滤掉这些行，这就是你遇到报错的原因。

可行的解决方案 #

方案1：文本格式读取后过滤解析

你提到的文本读取思路是可行的，调整代码逻辑后可以实现需求：

// 1. 读取所有行作为文本
val rawTextDF = spark.read.text("path/to/your/csv/file.csv")

// 2. 拆分每行并过滤出列数为3的行
val filteredRowsDF = rawTextDF
  .withColumn("split_columns", split(col("value"), ","))
  .filter(size(col("split_columns")) === 3)
  .select(col("value"))

// 3. 将过滤后的行解析为标准CSV DataFrame
val finalDF = spark.read
  .option("header", "true")
  .csv(filteredRowsDF)

方案2：全量读取后过滤裁剪

如果可以接受先读取所有列，再进行过滤，这种方式更简洁：

// 读取所有列（自动推断列数）
val tempDF = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .csv("path/to/your/csv/file.csv")

// 定义目标列数，过滤出列数匹配的行并选择前3列
val targetColumnCount = 3
val finalDF = tempDF
  .filter(size(split(concat_ws(",", tempDF.columns.map(col): _*), ",")) === targetColumnCount)
  .select(tempDF.columns.take(targetColumnCount).map(col): _*)

方案3：自定义解析逻辑（进阶）

如果需要更灵活的处理（比如处理复杂分隔符场景），可以使用mapPartitions逐行解析并过滤，但这种方式代码复杂度较高，适合特殊业务场景。

内容的提问来源于stack exchange，提问作者MOHAMMED SHEIK DAWOOD S J