为什么每个Spark任务都没有利用所有分配的内核？

Question

假设我每个执行器有36个核心，每个节点有一个执行器，以及3个节点，每个节点有48个可用核心。我注意到的基本要点是，当我将每个任务设置为使用1个内核（默认值）时，我对workers的CPU利用率约为70%，每个执行器将同时执行36个任务（正如我所预期的那样）。然而，当我将配置更改为每个任务有6个内核时（conf spark.task.cpus=6），每个执行器一次会减少到6个任务（如预期的那样），但我的CPU利用率也会下降到10%以下（意外的）。我假设Spark知道如何在6核上并行化工作负载。

重要的实现细节是，我正在数据帧的一列上运行一个UDF函数，并将结果作为新列附加到该数据帧上。此UDF函数使用一个瞬态对象，该对象提供我正在使用的机器学习算法。此UDF函数不是聚合或合并操作的一部分，它只是对列执行的映射操作，如下所示：

def myUdf = udf { ... }

val resultSet = myUdf(dataFrame.col("originalCol"))
val dataFrameWithResults = dataFrame.withColumn("originalColMetric", resultSet)

我原以为Spark会执行6个myUdf，一次处理6条记录，每个核心一条，但事实似乎并非如此。有没有办法解决这个问题（无需向Spark项目提交PR），或者至少有人能解释为什么会发生这种情况？

预料到这个问题，我正在尝试增加每个任务的内核数量，以减少每个执行程序所需的RAM量。在这种情况下，一次执行太多任务会成倍增加RAM的使用量。

Answer 1

<代码>火花。任务CPU是为每个任务分配的多个核心。当用户代码为多线程时，它用于将多个内核分配给单个任务。如果您的udf不使用多个线程（不会在一个函数调用中产生多个线程），那么内核就是浪费了。

一次处理6条记录

使用spark分配6个核心。任务CPU设置为1。如果要限制节点上的任务数，请减少每个节点提供的核心数。

实际上，Spark可以通过在每个任务之间拆分记录（根据分区）并确定每个执行者可以同时处理多少个任务，自行决定如何在多个记录上同时拆分映射UDF。但是，Spark不能自动将每个核心的工作分配给每个任务。要在每个任务中使用多个核心，需要编写UDF中的代码，以便在单个记录上并行该UDF中的计算，UDF中的代码将在每个任务的一次（顺序）在一条记录上执行。