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

Question

让我先说我对spark相对来说是新手，所以如果我说了一些没有意义的话，请纠正我。

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

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

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

当我查看历史服务器时，我看到有两个阶段特别长（每个阶段几乎4分钟），在这些阶段中，有一个任务占用了超过90%的时间。我多次运行代码，尽管每次都部署在不同的数据节点上，但都是相同的任务，需要花费大量的时间。

会有什么问题？任何帮助都将不胜感激。

Answer 1

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

为此，假设您有两个数据副本。其中一个有两列，number和stringrep，其中number对于0-10000中的所有整数来说只是一行，而stringrep只是它的字符串表示形式，所以“One”、“two”、“three”等等。我们将此numtostring称为numtostring

另一个DataFrame有一些键列要连接到numtostring中的number名为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的数据（具有种类==1)发送到一个执行器，%5的数据（具有种类==2)发送到另一个执行器，而剩下的%5则会在其余的执行器之间涂抹，从而使两个执行器具有巨大的分区，其余的执行器具有非常小的分区。

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

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

...它将向每个执行器发送NumtoString。假设ourdata事先被均匀分区，那么数据应该在执行器之间保持均匀分区。这可能不是你的问题，但听起来确实很像我们遇到的问题。希望有帮助！