PySpark映射报错：split后索引越界问题求助

核心问题定位 #

索引越界的直接原因是部分行经过row.split(",")拆分后，元素数量不等于预期的6个，导致访问row[1]、row[3]等索引时触发异常。

第一步：定位异常行 #

先修改代码找出所有格式异常的行，明确问题来源：

rdd3 = sc.textFile('hdfs://path/data.csv')
header3 = rdd3.first()

# 筛选出拆分后长度不等于6的行
bad_rows = rdd3.filter(lambda line: line != header3)\
               .map(lambda line: (line, len(line.split(","))))\
               .filter(lambda item: item[1] != 6)\
               .collect()

print("异常行详情：", bad_rows)

运行后就能看到哪些行不符合格式，常见的异常原因包括：空行、不完整的截断行、字段内包含逗号、header过滤不彻底。

针对不同异常的解决方案 #

1. 存在空行/不完整行 #

直接过滤掉拆分后长度不符合的行，在映射前增加过滤逻辑：

rdd3 = sc.textFile('hdfs://path/data.csv')
header3 = rdd3.first()
rdd3 = rdd3.filter(lambda line: line != header3 and line.strip() != "")\
       .map(lambda row: row.split(","))\
       .filter(lambda row: len(row) == 6)\  # 只保留符合字段数的行
       .map(lambda row: (row[0],row[1],row[3],row[5])) \
       .collect()

2. Header过滤不彻底 #

如果存在重复header行，或者line != header3匹配失效（比如大小写差异、前后空格），改用更精准的判断方式：

rdd3 = sc.textFile('hdfs://path/data.csv')
header3 = rdd3.first()
# 通过第一个字段判断是否为header
rdd3 = rdd3.filter(lambda line: line.split(",")[0] != "X")\
       .map(lambda row: row.split(","))\
       .filter(lambda row: len(row) == 6)\
       .map(lambda row: (row[0],row[1],row[3],row[5])) \
       .collect()

3. 字段内包含逗号（CSV标准格式场景） #

如果数据中存在带逗号的字段（比如地址字段被引号包裹，如"123 Main St, Apt 4B"），split(",")会错误拆分这类字段，此时需要用标准CSV解析器处理：

import csv
from io import StringIO

def parse_csv_line(line):
    # 使用csv模块解析单行CSV
    reader = csv.reader(StringIO(line))
    return next(reader)

rdd3 = sc.textFile('hdfs://path/data.csv')
header3 = rdd3.first()
rdd3 = rdd3.filter(lambda line: line != header3)\
       .map(parse_csv_line)\
       .filter(lambda row: len(row) == 6)\
       .map(lambda row: (row[0],row[1],row[3],row[5])) \
       .collect()

csv模块会自动识别带引号的字段，避免错误拆分。

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