(PySpark)reduceByKey之后的嵌套列表

python apache-spark

9618 观看

2回复

467 作者的声誉

我确信这很简单,但我找不到与此有关的任何内容。

我的代码很简单:

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

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

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

等等。所以,有时我得到一个平坦的值(如果它是单一的)。有时 - 嵌套列表可能非常非常深(在我的简单测试数据上它是3级深度)。

我尝试在源代码中搜索“flat”之类的东西 - 但是只找到了flatMap方法(据我所知)并不是我需要的方法。

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

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

itertools.chain停止在第一个找到的非可迭代值上展开。换句话说,它仍然需要一些编码(前一段)。

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

谢谢

作者: Spaceman 的来源 发布者: 2014 年 1 月 12 日

回应 (2)


5

10610 作者的声誉

决定

这里的问题是你的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

作者: Josh Rosen 发布者: 13.01.2014 07:38

0

21 作者的声誉

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

key1 [value1]
key2 [value2, value3]
key3 [value4, value5, value6]
作者: tianzi 发布者: 02.01.2016 07:44
32x32