Spark 执行器在加入时内存不足

Question

您好，我正在使用 spark Mllib 并在 1M 数据集和 1k 数据集之间进行 approxSimilarityJoin。
当我这样做时，我会广播 1k 的。
我看到的是，工作在倒数第二个任务中停止。
所有的执行者都死了，但有一个会持续运行很长时间，直到内存不足。
我检查了神经节，它显示内存一直在上升，直到达到极限

磁盘空间一直在减少，直到完成：

我调用的操作是 write，但它对 count 的作用相同。
现在我想知道：是否有可能集群中的所有分区都汇聚到一个节点上并造成这个瓶颈？

这是我的代码 sn-p：

var dfW = cookesWb.withColumn("n", monotonically_increasing_id())
    var bunchDf = dfW.filter(col("n").geq(0) && col("n").lt(1000000) )
    bunchDf.repartition(3000)
    model.
    approxSimilarityJoin(bunchDf,broadcast(cookesNextLimited),80,"EuclideanDistance").
    withColumn("min_distance", min(col("EuclideanDistance")).over(Window.partitionBy(col("datasetA.uid")))
                  ).
        filter(col("EuclideanDistance") === col("min_distance")).
      select(col("datasetA.uid").alias("weboId"),
        col("datasetB.nextploraId").alias("nextId"),
        col("EuclideanDistance")).write.format("parquet").mode("overwrite").save("approxJoin.parquet")

Answer 1

我会尽量回答。 在 Spark 中，有些东西被称为 shuffle 操作，它们完全按照您的想法进行操作，经过一些计算后，它们会将所有信息传输到单个节点。 如果你仔细想想，如果不把所有数据放在最后一个节点中，这些操作就没有其他方法可以工作。

连接操作示例： 您必须在 2 个不同的节点上进行分区

partition 1:
s, 1
partition 2:
s, k

并且您想通过 s 加入。 如果您没有在一台机器上获得两行，则无法计算出它们需要连接。

count 和 reduce 以及更多操作也是如此。 您可以阅读有关 shuffle 操作的信息或询问我是否需要更多说明。

一个可能的解决方案是： 而不是只将数据保存在内存中，您可以使用类似的东西：

dfW.persist(StorageLevel.MEMORY_AND_DISK_SER)

持久化还有其他选项，但它的作用基本上是将分区和数据不仅保存在内存中，而且以序列化的方式保存在磁盘中以节省空间。