PySpark使用reduce计算标准差抛出OverflowError问题排查

你的reduce逻辑完全搞错了，这才导致溢出错误：

reduce的工作逻辑是：

• 第一次调用函数时，传入RDD的前两个元素作为参数
• 之后每次调用，都会把上一次函数的返回值作为第一个参数，再传入RDD的下一个元素作为第二个参数

而你写的partial函数里，把第一个参数x也当成原始数据去计算(x - mean) ** 2，这就导致：第一次计算的是(元素1-mean)² + (元素2-mean)²，第二次就把这个和当成x，计算(和-mean)² + (元素3-mean)²——这个数值会指数级暴涨，很快超出Python浮点数的范围，触发OverflowError。

而你用collect()之后循环累加的逻辑是对的：每次只计算单个原始元素的平方差，然后加到总和里，数值是线性增长的，所以不会溢出。

修正后的两种写法 #

写法1：先转换再累加（最清晰） #

先把每个元素转换成对应的平方差，再用reduce做简单累加：

data = sc.textFile("data.csv") # Space-separated values
rdd  = data.map(lambda x: float(x.split(" ")[3])) # Only the fourth column
mean = rdd.mean() # High number: 1410000

# 先映射为平方差，再累加
part_sum = rdd.map(lambda x: (x - mean) ** 2).reduce(lambda a, b: a + b)
variance = part_sum / rdd.count() # 5.8 trillion
std_dev  = variance ** 0.5 # 和rdd.stdev()结果一致

写法2：修改reduce的累加逻辑 #

让函数的第一个参数作为累加器，只对第二个参数计算平方差：

data = sc.textFile("data.csv") # Same file
rdd  = data.map(lambda x: float(x.split(" ")[3])) # 4th column
mean = rdd.mean() # Same mean

def partial(acc, y):
    # acc是之前的累加总和，y是当前要处理的元素
    return acc + (y - mean) ** 2

# 用fold更稳妥，指定初始累加值为0.0
part_sum = rdd.fold(0.0, partial)
# 也可以用reduce，但要注意逻辑：lambda acc, y: acc + (y - mean)**2
variance = part_sum / rdd.count()
std_dev  = variance ** 0.5

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