PySpark执行.collect()/.count()报IndexError列表索引越界问题求助

问题根源 #

Spark的算子分为转换算子和行动算子，转换算子是惰性执行的，你代码中sc.textFile().map(parseRating)的逻辑在定义时不会实际执行，只有遇到行动算子才会触发全链路计算：

• show()默认仅拉取前20条数据计算展示，只要前20行数据都是符合::分隔、4个字段的规范，就不会触发报错
• collect()、count()需要扫描全量数据，只要存在任意一行格式不符合规范（比如空行、分隔符缺失、字段数不足4），parseRating函数中访问fields[3]就会触发数组越界错误

修复方案 #

方案1：给parseRating加容错逻辑 #

修改解析函数增加字段长度校验和异常捕获，同时过滤掉不符合格式的坏行：

def parseRating(line):
    """
    Parses a rating record in MovieLens format userId::movieId::rating::timestamp .
    """
    fields = line.strip().split("::")
    # 校验字段数量、同时捕获类型转换异常
    if len(fields) == 4:
        try:
            return int(fields[3]), int(fields[0]), int(fields[1]), float(fields[2])
        except (ValueError, IndexError):
            return None
    return None

创建RDD时增加过滤逻辑，丢弃无效行：

myRatingsRDD = sc.textFile("personalRatings.txt").map(parseRating).filter(lambda x: x is not None)
ratings = sc.textFile("ratings.dat").map(parseRating).filter(lambda x: x is not None)

方案2：直接用DataFrame API读取（更推荐） #

不需要自己手写解析逻辑，用Spark内置的CSV读取能力自动处理格式问题，性能和稳定性更好：

# 读取评分文件
df1 = spark.read.csv(
    "personalRatings.txt",
    sep="::",
    schema="userID INT, movieID INT, rating FLOAT, timestamp LONG"
)
df2 = spark.read.csv(
    "ratings.dat",
    sep="::",
    schema="userID INT, movieID INT, rating FLOAT, timestamp LONG"
)
# 直接丢弃格式错误的空行
df1 = df1.dropna()
df2 = df2.dropna()

额外修复点 #

你当前代码中myRatingsRDD、ratings的创建代码缩进在if __name__ == "__main__":块之外，脚本运行时可能出现上下文未初始化的问题，建议把所有业务逻辑缩进放到if __name__ == "__main__":的代码块内。

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