在火花中，如何同时运行几个“收集”

Question

我是 Spark 的新手，我不确定我是否理解它很好地分配作业的方式。

我有以下代码：

c1 = dict(smallrdd1.collect())
bigrdd1 = bigrdd1.filter(lambda val: c1[val])

c2 = dict(smallrdd2.collect())
bigrdd2 = bigrdd2.filter(lambda val: c2[val])

如果我理解得很好，这两个collect 将在主作业上运行，而不是同时运行，这会减慢整个过程。如何让 spark 在单独的作业中、在单独的节点上执行收集和过滤？

编辑

我认为我的问题不是很清楚。我会尽量让它更具体。

我想要实现的是对面向列的数据进行有效的星图连接。也就是说，我有几个维度表，每个维度表都有几百万个键值。当存储为原始 python dict 或 scala Map 时，它们每个都是几百 Mb。这足够小，可以在每个节点上单独放入内存中，但它们加在一起构成了几个 Gb，并且无法放入内存中。

另一方面，对于这些小维度列中的每一个，我都有一个大的事实列，它有几十亿行，并且肯定不适合内存。

这个想法是管理集群内的数据位置，以便每个大表位于不同的节点上。然后启动几个任务，每个任务负责一个join。然后从磁盘加载维度表并同时创建哈希图，然后将哈希图“广播”到那些需要它的任务。然后，将大事实列与维度表连接起来。

Answer 1

因此，collect 以及 take 和 first 通常用于测试和调试。正如@Shankar 所写，collect 会将所有数据拉到驱动程序中，这对于测试期间使用的小数据集来说很好，但在处理大量数据集时就不行了，因为它会用 OOM 杀死你的驱动程序。

现在，您说要加入，所以只需使用join：

val joined = bigrdd.join(smallrdd, bigrdd.col("id") === smallrdd.col("id"))

（这是 Scala 语法，但我相信你明白了）

现在数据已加入您的工作人员，您可以继续使用过滤器、地图和其他转换 :)

Answer 2

所以，我认为对我的问题的简短回答是不可能。

从我得到的答案来看，我认为这根本是不可能用 spark 做的。我们唯一的可能是在驱动程序中创建所有的hasmap，或者使用传统的spark join。

驱动程序内部

顺序加载小维表，然后将它们广播给所有执行器（这是顺序的，并且会导致 Gbs 的无用数据移动）：

c1 = sc.broadcast(smallrdd1.collectAsMap())
bigrdd1 = bigrdd1.filter(lambda val: c1.value[val])

c2 = sc.broadcast(smallrdd2.collectAsMap())
bigrdd2 = bigrdd2.filter(lambda val: c1.value[val])

关于执行者

不在驱动程序上计算任何东西，但计算连接的速度较慢，因为它涉及对大型数据集的 shuffle 操作。

bigrdd1 = bigrdd1 \
              .map(lambda (bigk,bigv):(bigv,bigk)) \
              .join(smallrdd1) \
              .map(lambda (bigv,(bigk,smallv)):(bigk,bigv))
bigrdd2 = bigrdd2 \
              .map(lambda (bigk,bigv):(bigv,bigk)) \
              .join(smallrdd2) \
              .map(lambda (bigv,(bigk,smallv)):(bigk,bigv))