Spark 性能：本地比集群快（非常不均匀的执行器负载）

Question

首先让我说我对火花比较陌生，所以如果我说的东西没有意义，请纠正我。

总结问题，不管我做什么，在某些阶段，一个执行器完成所有计算，这使得集群执行比本地的单处理器执行慢。

全文： 我编写了一个 spark 1.6 应用程序，它由一系列映射、过滤器、连接和一个简短的 graphx 部分组成。该应用程序仅使用一个数据源 - csv 文件。出于开发目的，我创建了一个模型数据集，包含 100 000 行，7MB，所有字段都具有均匀分布的随机数据（文件中也是随机排序）。连接是 PairRDD 上各个字段的自内连接（数据集有重复的键，每个键有约 200 个重复项，模仿真实数据），导致键内的笛卡尔积。然后我对连接的结果执行一些映射和过滤操作，将其存储为一些自定义类对象的 RDD，并将所有内容保存为图。

我在笔记本电脑上开发代码并运行它，大约需要 5 分钟（Windows 机器，本地文件）。令我惊讶的是，当我将 jar 部署到集群（主纱线、集群模式、HDFS 中的 csv 文件）并提交时，代码执行了 8 分钟。 我用较小的数据进行了相同的实验，结果在本地是 40 秒，在集群上是 1.1 分钟。

当我查看历史服务器时，我发现 2 个阶段特别长（每个阶段几乎 4 分钟），并且在这些阶段中，有一个任务需要超过 90% 的时间。我多次运行代码，即使每次都部署在不同的数据节点上，它总是花费大量时间的同一个任务。

令我惊讶的是，当我打开执行器时，我看到一个执行器几乎完成了所有工作（就花费的时间而言）并执行了大多数工作。在提供的屏幕截图中，第二个最“活跃”的执行者有 50 个任务，但情况并非总是如此——在不同的提交中，第二个最繁忙的执行者有 15 个任务，而领先的一个有 95 个任务。

此外，我看到 3.9 分钟的时间用于计算（第二个屏幕截图），这对 map 后不久的连接数据来说是最重的。我认为，数据可能不会被平均划分，一个执行者必须执行所有计算。因此，我尝试在加入之前（类似的执行时间）或加入之后（执行更慢）手动（使用 .partitionBy(new HashPartitioner(40))）对 pairRdd 进行分配。

可能是什么问题？任何帮助将不胜感激。

Answer 1

如果不查看您的查询并了解您的数据集，很难判断，我猜您也没有包含它，因为它非常复杂或敏感？所以这有点像在黑暗中拍摄，但这看起来很像我们在工作中处理的问题。我对正在发生的事情的粗略猜测是，在您的一次连接期间，您有一个具有高基数但分布非常不均匀的键空间。在我们的案例中，我们加入了网络流量来源，虽然我们有数千种可能的流量来源，但绝大多数流量都来自少数几个。这在我们加入时引起了问题。密钥将在执行者之间平均分配，但是由于可能 95% 的数据共享可能是 3 或 4 个密钥，因此只有极少数的执行者完成了大部分工作。当您发现一个存在此问题的连接时，要做的是选择两个数据集中较小的一个，并显式执行广播连接。 （Spark 通常会尝试这样做，但它并不总是能够完美地判断何时应该这样做。）

为此，假设您有两个 DataFrame。其中一个有两列，number 和 stringRep 其中数字只是来自0-10000 的所有整数的一行，stringRep 只是一个字符串表示，所以“一”、“二”、“三"等。我们称之为numToString

另一个 DataFrame 有一些关键列要加入到 number 中的 numToString 称为 kind、一些其他不相关的数据和 100,000,000 行。我们将此 DataFrame 称为ourData。然后假设ourData 中 100,000,000 行的分布是 90% 有kind == 1，5% 有kind == 2，剩下的 5% 在剩余的 99,998 个数字中分布得相当均匀。当您执行以下代码时：

val numToString: DataFrame = loadNumToString()
val ourData: DataFrame = loadOurCode()

val joined = ourData.join(numToString).where(ourData("kind") === numToString("number"))

...Spark 很可能会将 %90 的数据（具有 kind == 1 的数据）发送到一个执行器，将 %5 的数据（具有 kind == 2 的数据）发送到另一个执行器，并且剩下的 %5 涂抹在其余部分上，留下两个执行器的分区很大，其余的分区非常小。

正如我之前提到的，解决此问题的方法是显式执行广播连接。这样做是获取一个 DataFrame 并将其完全分发到每个节点。所以你会这样做：

val joined = ourData.join(broadcast(numToString)).where(ourData("kind") === numToString("number"))

...这会将numToString 发送给每个执行者。假设ourData 事先被均匀分区，数据应该在执行器之间保持均匀分区。这可能不是您的问题，但听起来确实很像我们遇到的问题。希望对您有所帮助！

更多关于广播连接的信息可以在这里找到： https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-joins-broadcast.html