GCP 上的 PySpark PandasUDF - 内存分配

Question

我正在使用 pandas udf 在 Dataproc (Spark) 中的 GCP 上训练许多 ML 模型。主要思想是我有一个分组变量，它代表我的数据框中的各种数据集，我运行如下：

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def test_train(grp_df):
    
  #train model on grp_df
  #evaluate model 
  #return metrics on 
 
    return (metrics)

result=df.groupBy('group_id').apply(test_train)

这工作正常，除非我使用非采样数据，其中返回的错误似乎与内存问题有关。这些消息是神秘的（对我来说），但如果我对它运行的数据进行采样，如果我不这样做，它就会失败。错误消息是这样的：

OSError：读取超出范围（偏移量 = 631044336，大小 = 69873416） 文件大小为 573373864

或

容器因超出内存限制而被 YARN 杀死。 24.5 GB 的 24 已使用 GB 物理内存。考虑提升 spark.yarn.executor.memoryOverhead 或禁用 由于 YARN-4714，yarn.nodemanager.vmem-check-enabled。

我的问题是如何在集群中设置内存以使其正常工作？

我了解每组数据和正在运行的进程都需要完全适合执行程序的内存。我目前有一个包含以下内容的 4-worker 集群：

如果我认为最大group_id中的最大数据大小需要150GB的内存，看来我真的需要每台机器一次操作一个group_id。与拥有单个工作人员或 VM 相比，我的速度至少提高了 4 倍。

如果我执行以下操作，这实际上是否为每台机器创建了 1 个执行程序，可以访问所有内核减去 1 和 180 GB 的内存？因此，如果理论上最大的数据组可以在具有这么多 RAM 的单个 VM 上工作，那么这个过程应该可以工作吗？

spark = SparkSession.builder \
  .appName('test') \
  .config('spark.executor.memory', '180g') \
  .config('spark.executor.cores', '63') \
  .config('spark.executor.instances', '1') \
  .getOrCreate() 

Answer 1

让我们把答案分成三部分：

1. 执行者数量
2. GroupBy 操作
3. 你的执行者记忆

执行者数量

直接来自Spark docs：

 spark.executor.instances

 Initial number of executors to run if dynamic allocation is enabled.
 If `--num-executors` (or `spark.executor.instances`) is set and larger
 than this value, it will be used as the initial number of executors.

所以，不。 你只会得到一个执行器，除非启用动态分配，否则它不会向上扩展。

您可以通过配置spark.executor.instances 手动增加此类执行器的数量，或者通过启用动态执行器分配来设置基于工作负载的自动扩展。

要启用动态分配，您还必须启用 shuffle 服务，该服务允许您安全地删除执行程序。这可以通过设置两个配置来完成：

1. spark.shuffle.service.enabled 到 true。默认为 false。
2. spark.dynamicAllocation.enabled 到 true。默认为 false。

分组方式

我观察到 group_by 在 Spark 中使用哈希聚合完成，这意味着给定 x 分区数，并且唯一的 group_by 值大于 x，多个 group by 值将位于同一分区中。

例如，假设 group_by 列中的两个唯一值是 a1 和 a2，总行大小分别为 100GiB 和 150GiB。

如果它们属于单独的分区，您的应用程序将运行良好，因为每个分区都将适合执行程序内存 (180GiB)，这是内存中处理所必需的，如果它们不适合，其余的将溢出到磁盘剩余的记忆。但是，如果它们属于同一个分区，您的分区将不适合执行程序内存（180GiB

在这种情况下，配置spark.default.parallelism 以将您的数据分布在相当多的分区上或应用加盐或其他技术来消除数据偏斜非常有用。

如果你的数据不是太倾斜，你说得对，只要你的 executor 可以处理最大的 groupby 值，它应该可以工作，因为你的数据将被均匀分区，并且发生上述情况的机会很少。

还有一点需要注意的是，由于你使用的是group_by，需要数据shuffle，所以你也应该开启shuffle服务。如果没有 shuffle 服务，每个 executor 必须在处理 shuffle 请求的同时完成自己的工作。

执行器内存

Spark 中的总执行器内存（实际执行器容器大小）是通过将分配给容器的执行器内存与分配的memoryOverhead 相加来确定的。 memoryOverhead 考虑了 VM 开销、内部字符串、其他本机开销等。所以，

Total executor memory = (spark.executor.memory + spark.executor.memoryOverhead)
spark.executor.memoryOverhead = max(executorMemory*0.10, 384 MiB)

基于此，您可以根据您的数据配置执行器以具有适当的大小。 因此，当您将spark.executor.memory 设置为180GiB 时，实际启动的执行器应该在198GiB 左右。

Answer 2

要解决纱线开销问题，您可以通过添加 .config('spark.yarn.executor.memoryOverhead','30g') 来增加纱线开销内存，并且为了获得最大并行度，建议将核心数保持为 5，因为您可以增加执行程序的数量。

spark = SparkSession.builder \
  .appName('test') \
  .config('spark.executor.memory', '18g') \
  .config('spark.executor.cores', '5') \
  .config('spark.executor.instances', '12') \
  .getOrCreate()  

# or use dynamic resource allocation refer below config 

spark = SparkSession.builder \
    .appName('test') \
   .config('spark.shuffle.service.enabled':'true')\
   .config('spark.dynamicAllocation.enabled':'true')\
   .getOrCreate()