python - Spark 广播与加入

Question

我有一个大的 RDD (rdd_1) 和它的一个过滤子集 (rdd_2)。我想在不同的字段上加入 rdd_1 和 rdd_2。

假设记录的格式为 {'first_name':<>, 'last_name':<>}。我们想要找到与所有“jack”具有相同姓氏的所有名字。

names = sc.textfile(RAW_DATA)
jack = names.filter(lambda v: v['first_name'] == 'jack')

选项 1

jack_last_names = jack.map(operator.itergetter('last_name').distinct().collect()
last_names_bc = sc.broadcast(set(jack_last_names))
final = names.filter(lambda v:v['last_name'] in last_names_bc.value)

目前，我广播rdd_2并通过它过滤rdd_1。问题是，为了广播 rdd_2，我必须首先在驱动程序上收集()它，这会导致驱动程序内存不足。

有没有一种方法可以在不先在驱动程序上收集()的情况下广播 RDD？

选项 2

final = jack.keyBy(operator.itemgetter('last_name').join(names.keyBy(operator.itemgetter('last_name')

我的另一个选择是 rdd_1.join(rdd_2) 但 rdd_1 太大而无法洗牌。

当我们运行 rdd_1.join(rdd_2) 时，rdd_1 和 rdd_2 都会进行哈希分区和洗牌吗？

谢谢!

Answer 1

Is there a way to broadcast an RDD without first collect()ing it on the driver?

不，不存在，即使存在也不能解决您的问题。

• 无法执行嵌套操作或转换
• 如果您可以创建一个不带集合的本地广播变量，您将面临同样的问题，但在工作人员身上

When we run rdd_1.join(rdd_2) do both rdd_1 and rdd_2 get hash partitioned and shuffled?

从技术上讲，在 PySpark 中，它需要 union 后跟 groupByKey，因此这意味着所有数据都必须进行混洗。

实际上，我会简单地接受重新洗牌的成本。一般来说，编写任何复杂的应用程序并完全避免改组是不可能的。而且也不比broadcasting a similar amount of data贵。甚至通过复制将数据复制到分布式文件系统。