GCP上的PySpark PandasUDF-内存分配

问题描述

我正在使用熊猫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.memory开销或禁用由于YARN-4714，启用yarn.nodemanager.vmem-check。

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

我了解到，每组数据和正在运行的流程都必须完全适合执行者的内存。我目前有一个包含以下内容的4人集群：

如果我认为最大的group_id中的最大数据大小需要150GB内存，看来我真的需要每台机器一次在一个group_id上运行。与只有一个工作程序或VM相比，我至少获得了4倍的速度。

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

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

解决方法

让我们将答案分为三个部分：

执行人数量
GroupBy操作
您的执行者记忆力

执行人数量

直接从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来手动增加此类执行程序的数量，或者通过启用动态执行程序分配来设置基于工作负载的自动扩展。

要启用动态分配，还必须启用随机播放服务，该服务可以安全删除执行程序。可以通过设置两个配置来完成：

spark.shuffle.service.enabled至true。默认值为false。
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配置为在合理数量的分区上分布数据或应用盐析或其他技术来消除数据偏斜很有用。

如果您的数据不太偏斜，那么您可以正确地说，只要您的执行者可以处理最大的groupby值，它就可以正常工作，因为您的数据将被均匀地分区，并且发生上述情况的机会很少。 / p>

要注意的另一点是，由于您使用的是group_by，需要进行数据混洗，因此您还应该打开混洗服务。如果没有洗牌服务，则每个执行者都必须同时执行洗牌请求和自己的工作。

执行者记忆力

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左右。

要解决纱线开销问题，可以通过添加.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()

apache-spark google-cloud-dataproc google-cloud-storage pyarrow pyspark