用于巨大列表的Spark FlatMap函数

Question

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

我的问题是:如果这个列表太大而你的记忆无法处理,会发生什么？

我目前还没有实现这个,在我重写MapReduce软件之前应该解决这个问题,这个软件可以通过context.write()在我想要的算法中放置任何地方来轻松解决这个问题.(单个映射器的输出可能很容易大量的千兆字节.

如果您感兴趣:映射器会进行某种字数统计,但事实上它会生成所有可能的子字符串,以及与文本匹配的各种正则表达式.(生物信息学用例)

Answer 1

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

不,它不必返回列表.在实践中,您可以轻松使用惰性序列.在查看Scala RDD.flatMap签名时,可能更容易发现:

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

由于TraversableOnceinclude的子类SeqView或者Stream你可以使用惰性序列而不是a 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)