apache-spark - 从驱动程序同时运行多个 Spark 作业

Question

想象一下，我们有 3 个客户，我们希望同时为每个客户做一些相同的工作。

def doSparkJob(customerId: String) = {
  spark
    .read.json(s"$customerId/file.json")
    .map(...)
    .reduceByKey(...)
    .write
    .partitionBy("id")
    .parquet("output/")
}

我们同时这样做(来自 Spark 驱动程序):

val jobs: Future[(Unit, Unit, Unit)] = for {
  f1 <- Future { doSparkJob("customer1") }
  f2 <- Future { doSparkJob("customer1") }
  f3 <- Future { doSparkJob("customer1") }
} yield (f1, f2, f3)

Await.ready(jobs, 5.hours)

我是否正确理解这是不好的方法？许多 Spark 作业会从执行器中推出彼此的上下文，并且会出现许多将数据溢出到磁盘的情况。 spark 将如何管理并行作业的执行任务？当我们有来自一个驱动程序的 3 个并发作业并且只有 3 个具有一个内核的执行程序时，shuffle 会如何出现。

我想，一个好的方法应该是这样的:
我们按客户一起读取所有客户 groupByKey 的所有数据，然后做我们想做的事情。

Answer 1

Do I understand correctly that this is bad approach?

不必要。很大程度上取决于上下文，Spark 实现了它自己的一套 AsyncRDDActions解决这样的场景(尽管没有 Dataset 等效项)。

在最简单的情况下，使用静态分配，由于缺乏资源，Spark 很可能只会按顺序调度所有作业。除非另有配置，否则这是所描述配置最可能的结果。请记住，Spark 可以使用应用程序内调度和 FAIR 调度程序在多个并发作业之间共享有限的资源。见 Scheduling Within an Application .

如果资源量足以同时启动多个作业，则单个作业之间可能存在竞争，尤其是 IO 和内存密集型作业。如果所有作业使用相同的资源(尤其是数据库)，则 Spark 可能会导致节流和后续故障或超时。运行多个作业的一个不太严重的影响可能是增加缓存驱逐。

总的来说，在顺序执行和并发执行之间进行选择时需要考虑多个因素，包括但不限于可用资源(Spark 集群和外部服务)、API 的选择(RDD 往往比 SQL 更贪婪，因此需要一些低级别的管理)和运营商的选择。即使作业是按顺序进行的，您仍可能决定使用异步来提高驱动程序利用率并减少延迟。这对于 Spark SQL 和复杂的执行计划(Spark SQL 中的常见瓶颈)特别有用。这样 Spark 可以处理新的执行计划，同时执行其他作业。