python - (PySpark) reduceByKey 后的嵌套列表

Question

我确定这是非常简单的事情，但我没有找到与此相关的任何内容。

我的代码很简单:

... 
stream = stream.map(mapper) 
stream = stream.reduceByKey(reducer) 
... 

没什么特别的。输出如下所示:

... 
key1  value1 
key2  [value2, value3] 
key3  [[value4, value5], value6] 
... 

等等。所以，有时我得到一个固定值(如果它是单一的)。有时 - 嵌套列表可能非常非常深(在我的简单测试数据中它有 3 层深)。

我尝试通过源代码搜索类似“平面”的东西 - 但只找到了 flatMap 方法，这(据我所知)不是我需要的。

我不知道为什么这些列表是嵌套的。我的猜测是它们由不同的进程( worker ？)处理，然后在没有展平的情况下连接在一起。

当然，我可以用 Python 编写代码来展开该列表并将其展平。但我相信这不是正常情况——我认为几乎每个人都需要平稳的输出。

itertools.chain 在第一个发现的不可迭代值上停止展开。换句话说，它仍然需要一些编码(上一段)。

那么 - 如何使用 PySpark 的 native 方法来展平列表？

谢谢

Answer 1

这里的问题是你的 reduce 函数。对于每个键，reduceByKey 使用值对调用您的 reduce 函数，并期望它产生相同类型的组合值。

例如，假设我要执行字数统计操作。首先，我可以将每个单词映射到一个 (word, 1) 对，然后我可以 reduceByKey(lambda x, y: x + y) 总结每个字。最后，我得到了 (word, count) 对的 RDD。

这是来自 PySpark API Documentation 的示例:

>>> from operator import add
>>> rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
>>> sorted(rdd.reduceByKey(add).collect())
[('a', 2), ('b', 1)]

要理解为什么您的示例不起作用，您可以想象像这样应用 reduce 函数:

reduce(reduce(reduce(firstValue, secondValue), thirdValue), fourthValue) ...

根据您的 reduce 函数，听起来您可能正在尝试实现内置的 groupByKey操作，它将每个键与其值列表分组。

另外，看看 combineByKey ，reduceByKey() 的泛化，它允许 reduce 函数的输入和输出类型不同(reduceByKey 是 implemented combineByKey)