Pyspark udf 内存利用率高

Posted 2023-04-15

【中文标题】Pyspark udf 内存利用率高

【英文标题】：Pyspark udf high memory utilization

【发布时间】：2017-06-18 08:16:14

【问题描述】：

我正在使用用 python 编写的 UDF 来更改数字的基数。

所以我读取了 parquet 文件并写入 parquet 文件并应用 UDF。 这是我运行的行：

input_df.withColumn("origin_base", convert_2_dest_base(input_df.origin_base)).write.mode('overwrite').parquet(destination_path)

这种转换使 spark 使用大量内存，我收到这种警告：

17/06/18 08:05:39 WARN TaskSetManager：在 4.0 阶段丢失任务 40.0（TID 183，ip-10-100-5-196.ec2.internal，执行程序 19）：ExecutorLostFailure（执行程序 19 退出导致由正在运行的任务之一）原因：容器因超出内存限制而被 YARN 杀死。使用了 4.4 GB 的 4.4 GB 物理内存。考虑提升 spark.yarn.executor.memoryOverhead。

最后还是失败了。

UDF 不是正确的方法吗？为什么会消耗这么多内存？

【问题讨论】：

如果不分享一个完全可重复的例子，很难说出什么是正确的方法。

【参考方案1】：

对于 pyspark，数据在 Python 中处理并在 JVM 中缓存/洗牌。如果您使用的是内置 Python API，那么 scala 的性能不会有太大差异。见python vs scala performance

当你使用 udf 时，由于你的本地定义函数没有注册到本地 JVM 结构中，因此不能通过简单的 java API 调用来实现，它必须序列化/反序列化到 Python 工作者。然后数据将在 Python worker 中处理并序列化/反序列化回 JVM。

Python worker 现在需要在堆外内存中处理序列化的数据，它会消耗巨大的堆外内存，因此经常会导致内存开销。

性能方面，serialization 很慢，它通常是性能调优的关键。

【参考方案2】：

udf 函数对columns conversion 使用serialization 和deserialization 方法。这就是使用大量内存的原因。您可以查看spark functions 的替代方案。