用于大型列表的星火FlatMap函数

Question

我有一个非常基本的问题。Spark的flatMap函数允许每个输入发出0、1或更多输出。因此，您提供给flatMap的(lambda)函数应该返回一个列表。

我的问题是:如果这个列表太大，您的内存无法处理！？会发生什么？

我目前还没有实现这一点，在重写我的MapReduce软件之前，应该先解决这个问题，该软件可以通过将context.write()放在我想要的算法中的任何位置来轻松地处理这个问题。(一个映射器的输出可以很容易地产生很多千兆字节。

如果您感兴趣:映射器会进行某种类型的单词计数，但实际上，在生成所有可能的子字符串时，会生成与文本匹配的范围广泛的regex表达式。(生物信息学用例)

Answer 1

因此，您提供给flatMap的(lambda)函数应该返回一个列表。

不，它不需要退货单。在实践中，您可以轻松地使用惰性序列。当查看Scala RDD.flatMap签名时，可能更容易发现：

flatMap[U](f: (T) ⇒ TraversableOnce[U])

因为TraversableOnce的子类包括SeqView或Stream，所以可以使用惰性序列而不是List。例如：

val rdd = sc.parallelize("foo" :: "bar" :: Nil)
rdd.flatMap {x => (1 to 1000000000).view.map {
    _ => (x, scala.util.Random.nextLong)
}}

由于您已经提到lambda函数，我假设您使用的是PySpark。您可以做的最简单的事情是返回一个生成器而不是列表：

import numpy as np

rdd = sc.parallelize(["foo", "bar"])
rdd.flatMap(lambda x: ((x, np.random.randint(1000)) for _ in xrange(100000000)))

由于RDDs是延迟计算的，所以甚至有可能从flatMap返回无限序列。使用一点toolz电源：

from toolz.itertoolz import iterate
def inc(x):
    return x + 1

rdd.flatMap(lambda x: ((i, x) for i in iterate(inc, 0))).take(1)