在单次传递数据中使用 pyspark 找到 min/max

Question

我有一个包含大量数字列表的 RDD（文件中的行的长度），我想知道如何在单次传递数据时获得 min/max。

我知道关于 Min 和 Max 函数，但这需要两次传递。

Answer 1

试试这个：

>>> from pyspark.statcounter import StatCounter
>>> 
>>> rdd = sc.parallelize([9, -1, 0, 99, 0, -10])
>>> stats = rdd.aggregate(StatCounter(), StatCounter.merge, StatCounter.mergeStats)
>>> stats.minValue, stats.maxValue
(-10.0, 99.0)

Answer 2

这是一个使用累加器的可行但不够优雅的解决方案。不雅之处在于您必须事先定义 zero/initial 值，以免它们干扰数据：

from pyspark.accumulators import AccumulatorParam
class MinMaxAccumulatorParam(AccumulatorParam): 
    def zero(self, value): 
        return value
    def addInPlace(self, val1, val2): 
        return(min(val1[0],val2[0]), max(val1[1],val2[1]))

minmaxAccu = sc.accumulator([500,-500], MinMaxAccumulatorParam())

def g(x):
    global minmaxAccu
    minmaxAccu += (x,x)

rdd = sc.parallelize([1, 2, 3, 4, 5])

rdd.foreach(g)

In [149]: minmaxAccu.value
Out[149]: (1, 5)

在单次传递数据中使用 pyspark 找到 min/max

finding min/max with pyspark in single pass over data

python

apache-spark

rdd

pyspark