python - Pyspark MapReduce对象列表

Question

我有一个对象列表定义为:

class AD:
    def __init__(self, name=None, impressions=0):
        self.name = name
        self.impressions= impressions

现在，有一种方法可以使用map reduce方法来获取按名称汇总的最新AD列表，以及展示次数的总和。例如:

adList = []

ad1 = AD("BlackFriday",29)
ad2 = AD("BlackFriday",33)
ad3 = AD("CyberMonday",35)

adList .append(a1)
adList .append(a2)
adList .append(a3)

我知道我不必使用map reduce函数来获取结果。如果我要使用map reduce，如何获得以下结果:

newList[0].name = "BlackFriday"
newList[0].impressions = 62

newList[1].name = "CyberMonday"
newList[1].impressions = 35

谢谢您的帮助。

Answer 1

以下代码未经测试，因为我没有任何可用的环境

您的输入:

ad1 = AD("BlackFriday",29)
ad2 = AD("BlackFriday",33)
ad3 = AD("CyberMonday",35)

为了能够使用 map + reduce，您需要一些RDD(为此，您可以使用 parallelize)

rdd = sc.parallelize(list((ad1, ad2, ad3)))

然后，您将rdds map到 (name, impressions)元组，以便可以按key减少after

rddReduced = rdd.map(lambda a: (a.name, a.impressions)).reduceByKey(lambda x,y : x + y)

然后，您将 map返回到AD对象，并调用 collect以将结果作为列表获取

adList = rddReduced.map(lambda (n,i): AD(n,i)).collect()