如何通过spark中的Futures保证集群资源的有效利用

Question

我想在一个火花集群中并行运行多个火花SQL，这样我就可以广泛利用整个资源集群。我正在使用sqlContext.sql（查询）。

我在这里看到了一些示例代码，如下所示，

val parallelism = 10
val executor = Executors.newFixedThreadPool(parallelism)
val ec: ExecutionContext = ExecutionContext.fromExecutor(executor)
val tasks: Seq[String] = ???
val results: Seq[Future[Int]] = tasks.map(query => {
  Future{
    //spark stuff here
    0
  }(ec)
})
val allDone: Future[Seq[Int]] = Future.sequence(results)
//wait for results
Await.result(allDone, scala.concurrent.duration.Duration.Inf)
executor.shutdown //otherwise jvm will probably not exit 

据我所知，ExecutionContext计算机器中的可用内核（使用ForkJoinPool）并相应地进行并行处理。但是，如果我们考虑的是spark集群而不是单机，那么会发生什么呢？它如何保证完全的集群资源利用率呢。？

例如：如果我有一个10节点的集群，每个集群有4个核心，那么上面的代码如何保证使用40个核心。

编辑：-

假设有2个sql要执行，我们有两种方法可以做到这一点，

>

按顺序提交查询，以便仅在执行第一个查询后才能完成第二个查询。（因为sqlContext.sql（查询）是一个同步调用）

使用Futures并行提交这两个查询，以便在集群中独立并行执行这两个查询，前提是有足够的资源（在这两种情况下）。

我认为第二个更好，因为它使用了集群中可用的最大资源，如果第一个查询充分利用了资源，调度程序将等待作业完成（取决于策略），这在这种情况下是公平的。

但是正如user9613318提到的“增加池大小会使驱动程序饱和”那么我如何有效地控制线程以更好地利用资源。

Answer 1

并行性在这里的影响最小，额外的集群资源不会真正影响该方法<代码>未来（或线程）不是用来并行执行，而是用来避免阻塞执行。增加池大小只会使驱动程序饱和。

您真正应该关注的是应用程序调度池中的Spark以及窄分区数量的调整（如何在Spark-SQL中更改分区大小，partitionColumn、lowerBound、upperBound、numPartitions参数的含义是什么？）和广度（spark.sql.shuffle.partitions的最佳值是什么，或者在使用spark-sql时如何增加分区？）转换。

如果作业是完全独立的（代码结构表明），则最好使用自己的一组分配资源单独提交每个作业，并相应地配置集群调度池。