3

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

我的代码很简单：

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

没什么特别的。输出看起来像这样：

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

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

我尝试通过来源搜索“flat”之类的东西 - 但发现只有 flatMap 方法（据我了解）不是我需要的。

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

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

itertools.chain 在找到不可迭代的值时停止展开。换句话说，它仍然需要一些编码（上一段）。

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

谢谢

4

2 回答 2

5

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

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

这是PySpark API 文档中的一个示例：

>>> 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实现）combineByKey

于 2014-01-13T07:38:27.077 回答
1

或者，stream.groupByKey().mapValues(lambda x: list(x)).collect()给出

key1 [value1]
key2 [value2, value3]
key3 [value4, value5, value6]
于 2016-01-02T19:44:27.880 回答