使用Apache Sp将键值对缩减为键列表对

Question

我正在编写一个Spark应用程序，希望将一组键值对(K, V1), (K, V2), ..., (K, Vn)组合成一个键值多值对(K, [V1, V2, ..., Vn])。我觉得我应该能够使用reduceByKey函数来完成这项工作，它具有以下特点：

My_KMV = My_KV.reduce(lambda a, b: a.append([b]))

发生这种情况时，我得到的错误是：

'NoneType' object has no attribue 'append'.

我的键是整数，值V1，…，Vn是元组。我的目标是用键和值列表（元组）创建一对。

Answer 1

tl；dr如果确实需要这样的操作，请使用groupByKeyas suggestedby @MariusIon。与直接分组相比，本文提出的其他解决方案要么效率低下，要么至少是次优的。

带有列表连接的reduceByKey不是可接受的解决方案，因为：

• 需要初始化O（N）列表。
• 对一对列表应用+需要两个列表的完整副本（O（N））有效地将总体复杂性增加到O（N²）。
• 不解决由groupByKey引入的任何问题。必须洗牌的数据量和最终结构的大小是相同的。
• 与suggested by one of the answers不同，使用reduceByKey和groupByKey的实现在并行性级别上没有区别。

带list.extend的combineByKey是次优解决方案，因为：

• 在MergeValue中创建O（N）列表对象（这可以通过直接在新项上使用list.append来优化）。
• 如果使用list.append进行优化，则它完全等同于groupByKey的旧（Spark<；=1.3）实现，并且忽略Spark-3074引入的所有优化，这些优化允许对大于内存的结构进行外部（磁盘上）分组。

Answer 2

我有点晚了，但我的建议是：

>>> foo = sc.parallelize([(1, ('a','b')), (2, ('c','d')), (1, ('x','y'))])
>>> foo.map(lambda (x,y): (x, [y])).reduceByKey(lambda p,q: p+q).collect()
[(1, [('a', 'b'), ('x', 'y')]), (2, [('c', 'd')])]

Answer 3

映射和还原关键字

输入类型和输出类型reduce必须相同，因此如果要聚合列表，必须map列表的输入。然后你把这些列表合并成一个列表。

组合列表

您需要一个方法来将列表合并为一个列表。植物素提供一些methods to combine lists。

append修改第一个列表并始终返回None。

x = [1, 2, 3]
x.append([4, 5])
# x is [1, 2, 3, [4, 5]]

extend也这样做，但展开列表：

x = [1, 2, 3]
x.extend([4, 5])
# x is [1, 2, 3, 4, 5]

两个方法都返回None，但您需要一个返回组合列表的方法，因此只返回use the plus sign。

x = [1, 2, 3] + [4, 5]
# x is [1, 2, 3, 4, 5]

火花

file = spark.textFile("hdfs://...")
counts = file.flatMap(lambda line: line.split(" ")) \
         .map(lambda actor: (actor.split(",")[0], actor)) \ 

         # transform each value into a list
         .map(lambda nameTuple: (nameTuple[0], [ nameTuple[1] ])) \

         # combine lists: ([1,2,3] + [4,5]) becomes [1,2,3,4,5]
         .reduceByKey(lambda a, b: a + b)

---

组合键

也可以用combineByKey来解决这个问题，它在内部用于实现reduceByKey，但它更复杂，而且"using one of the specialized per-key combiners in Spark can be much faster"。您的用例对于上层解决方案来说已经足够简单了。

按键分组

也可以用groupByKey，but it reduces parallelization来解决这个问题，因此对于大数据集来说可能要慢得多。