如何在 Apache Spark 中实现任务的动态负载均衡

Question

我知道在 Spark 中我可以使用多个分区来拆分我的计算。如果说我可以将输入的 RDD 拆分为 1000 个分区并且我的机器数量为 100，Spark 会将计算拆分为 1000 个任务，并以某种智能方式将它们动态分配到我的 100 台机器中。

现在假设我最初只能将数据分成 2 个分区，但我仍然有 100 台机器。自然而然，我的 98 机器就闲置了。但是当我处理每个任务时，我可能会将其拆分为可能在不同机器上执行的子任务。它可以通过 queue 在纯 Java 中轻松实现，但我不确定在 Apache Spark 中攻击它的最佳方法是什么。

考虑以下 Java 伪代码：

BlockingQueue<Task> q = new LinkedBlockingQueue<Task>();
q.push(myInitialTask);
...
//On each thread:
while (!queue.isEmpty()) {
    Task nextTask = queue.take();
    List<Task> newTasks = process_task_and_split_to_sub_tasks(nextTask);
    queue.pushAll(newTasks);
} 

假设方法 'process_task_and_split_to_sub_tasks()' 可以将任何大型任务拆分为多个较小的任务，上述 Java 代码将使我所有的 100 个线程保持忙碌。

有没有办法在 Spark 中实现相同的功能，可以与其他工具结合使用吗？

更新：已经正确指出，攻击它的方法之一就是

1. 生成更细粒度的密钥和
2. 然后使用智能分区器将这些键分配给分区。

我想这是解决这个问题的“经典”方法，但它需要我能够正确估计每个键的工作量以正确分区它。如果我没有提前了解每个键的工作量的好方法怎么办？当我的大多数机器将保持空闲等待一些不幸的机器时，我可能会以非常不幸的分区告终。

示例：我们以一个简化的频繁项集挖掘为例。
假设我的文件包含字母从 a 到 j 的行（10 个字母），每行中的所有字母都按字母顺序排序并且没有重复，例如'abcf' 任务是找出所有行的 50% 中存在的所有字母组合。例如。如果很多行与模式 'ab.*f' 匹配，则输出将包含 {'a', 'b', 'f', 'ab', 'af', 'bf', 'abf'}。
实现它的一种方法是将所有以“a”开头的行发送到一个映射器（机器），所有以“b”开头的行发送到另一个等等。顺便说一句，这就是frequent pattern mining is implemented in Spark 的方式。现在假设我有 100 台机器（但只有 10 个字母）。然后我的 90 台机器将保持空闲状态。
使用更细粒度的密钥解决方案，我可以生成 10,000 个 4 字母前缀，然后根据每个前缀的估计工作量以某种方式对它们进行分区。但是我的分区可能是非常错误的：如果大多数行以'abcd'开头，那么所有工作都将由负责这个前缀（可能还有其他前缀）的机器完成，再次产生一个当我的大多数机器闲置等待一些不幸的机器时的情况。

在这种情况下，动态负载平衡将是这样的：收到以 'a' 开头的行的映射器可能希望进一步拆分其行 - 以 'ab 开头的行', 'ac', 'ad',... 然后将它们发送到其他 10 台机器，这些机器可能决定进一步将它们的工作拆分为更多任务。
我知道标准的 Apache Spark 没有开箱即用的答案，但我想知道是否有办法实现这一点。

Kafka（即上面的队列）+Spark Streaming 看起来很有前途，你认为我可以通过相对简单的方式使用这些工具来实现动态负载平衡吗？您能推荐其他工具吗？

Answer 1

现在假设我有 100 台机器（但只有 10 个字母）。收到以“a”开头的行的映射器可能希望进一步拆分其行 - 以“ab”、“ac”、“ad”等开头的行，然后将它们发送到其他 10 台机器。

这不是 Spark 的工作方式。 “映射器”（任务）大多不了解所有分布式上下文。在这个级别，无法访问SparkContext，我们不再拥有RDDs，只需输入本地Iterator 和要在其上执行的代码。它无法启动，也无法创建新任务。

同时，您的问题定义是人为的。要找到频繁的模式，您必须聚合数据，因此您需要 shuffle。在这些点上，对应于给定模式的记录必须被洗牌到同一台机器上。确保数据正确分布是Partitioner 的工作，这里真的没有“拆分”的地方。

Answer 2

Spark 自己的动态分配可以在一定程度上模拟您想要的，但是如果您需要具有低级控制的详细、高性能方法，那么 Spark 不适合您。对于初学者，您将无法动态拆分任务 - 您只能调整分配给应用程序的整体资源。

您应该考虑低级调度程序并从头开始实施您自己的解决方案。

Answer 3

要归档您的要求，您只需将数据从两个分区重新分区到您想要的任意数量的分区。

见https://spark.apache.org/docs/2.2.0/api/java/org/apache/spark/api/java/JavaPairRDD.html#repartition-int-

顺便说一句，火花流与您的问题无关。

请注意，并行度不仅取决于数据集的分区，还取决于我们的工作/算法。