嵌套组合RDD打印：<pyspark.resultiterable.resultiterable对象位于…>

Question

我在打印/保存RDD时遇到一些问题。下面是示例代码：

users_prods_joined = users_grpd.cogroup(prods_grpd) 

users_prods_joined_flattened = users_prods_joined.map(
   lambda (k, mapped): "{0} {1}".format(k,str(mapped)))

users_prods_joined_flattened.saveAsTextFile('users_prods_joined_flattened')

o/p是：

^{pr2}$

648018220 pyspark.resultiterable.ResultIterable object at 0x107cc5b50

100341214 pyspark.resultiterable.ResultIterable object at 0x107cc5b50

8429212 pyspark.resultiterable.ResultIterable object at 0x107cc5b50

。。。。。。在

我想知道如何获得实际值，而不是pyspark.resultiterable.ResultIterable object at 0x107cc5b50 .....

Answer 1

如果您还不想使用collect()，但仍然想扩展iteretarables，可以使用Harsha的示例执行以下操作：

>>> x = sc.parallelize([("foo", 1), ("bar", 4)])
>>> y = sc.parallelize([("foo", -1)])
>>> grp = x.cogroup(y).mapValues(lambda val: [i for e in val for i in e])

有点难看，但是在一行中，您得到的是所有分组值的单个列表，而不是Iterable

Answer 2

如果您查看位于https://www.safaribooksonline.com/library/view/learning-spark/9781449359034/ch04.html的cogroup文档，可以看到它返回RDD[（K，（Iterable[V]，Iterable[W]）]，其中Iterable[V]和Iterable[W]包含两个独立RDD的值。您必须对它们进行迭代，或者通过调用list（…）将它们转换成一个列表。在

我假设这样的代码可以工作（注：代码未测试）：

users_prods_joined.map(lambda (k, mapped): "{0} {1}".format(k,str(list(mapped[0]) + list(mapped[1])))) 

Answer 3

试试这个

[(x, tuple(map(list, y))) for x, y in sorted(list(x.cogroup(y).collect()))]

示例：

^{pr2}$

你应该得到想要的输出。。。在