Pyspark udf高内存使用率

Question

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

因此，我读了一个实木复合地板文件，并写入实木复合地板文件并应用UDF。 这里是我行运行：

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

转换，使得火花利用大量的内存，我得到这样的警告：

17/06/18 8时05分39秒WARN TaskSetManager：阶段4.0中丢失的任务40.0（TID 183，ip-10-100-5-196.ec2.internal，executor 19）：ExecutorLostFailure（执行程序19由其中一个正在运行的任务导致退出）原因：YARN为超出内存限制。 4.4 GB使用的4.4 GB物理内存。考虑提升spark.yarn.executor.memoryOverhead。

并最终失败。

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

Answer 1

对于pyspark，数据在Python中处理并在JVM中缓存/混排。如果您使用内置的Python API，那么在性能方面与scala不会有太大的区别。见python vs scala performance

当您使用UDF，因为本地定义的函数并不在本机JVM结构登记，因此无法通过简单的Java API调用来实现的，它必须是序列化/反序列化到Python工人。然后数据将在Python worker中处理并序列化/反序列化回JVM。

Python工作者现在需要在堆外存储器中处理序列化的数据，它消耗巨大的堆外存储器，因此它通常会导致memoryOverhead。

表现明智，serialization是缓慢的，它往往是性能调整的关键。

Answer 2

udf功能使用serialization和deserialization方法为columns conversion。这就是使用大量内存的原因。你可以看看spark functions的替代品。