如何在以下示例中使用 pyspark 折叠操作找到最大值？

Posted 2023-04-15

【中文标题】如何在以下示例中使用 pyspark 折叠操作找到最大值？

【英文标题】：How to find max using pyspark fold operation in following example?

【发布时间】：2018-11-09 10:00:54

【问题描述】：

我是 pyspark 和 python 的新手。所以，请帮我解决这个问题，我需要使用 fold 和使用 operator.gt 或编写我们自己的 lambda 函数来找到最大值。

以下由我编写的代码会引发无法索引 rdd 的错误。我明白了，但是如何传递和比较每个值 1、2、0、3 和 0 的值并找到最大值。 这里 0 是我的累加器值，1,2,0,3 是每次的当前值。 我正在尝试将用 scala 编写的解释折叠的程序转换为 python。 预期答案：('d', 3)

from pyspark import SparkContext
from operator import gt

def main():
    sc = SparkContext("local", "test")

    data = sc.parallelize([('a', 1), ('b', 2),('c', 0), ('d', 3)])

    #dummy = ('dummy', 0)

    maxVal = data.fold(0, lambda acc, a : gt(acc, a[1])).collect()

    print(maxVal)


if __name__ == '__main__':
    main()

【问题讨论】：

您了解什么是 lambda 表达式以及 fold 的工作原理吗？另一个提示：您需要的实际上不是一个简单的最大值（尽管它会涉及计算最大值），因为您不仅需要最大值，还需要包含该值的整行。

【参考方案1】：

使用neutral value（可以合并任意次数而不改变最终结果）适用于特定操作并匹配数据类型（函数应为(T, T) => T或使用Python约定@987654323 @)。使用 max 按值使用 float("-Inf") 和虚拟键是有意义的：

zero = (None, float("-Inf"))

减少使用max 和key：

from functools import partial
from operator import itemgetter

op = partial(max, key=itemgetter(1))

综合：

data.fold(zero, op)

('d', 3) 

当然在实践中你可以只使用RDD.max

data.max(key=itemgetter(1))

('d', 3)