apache-spark - Pyspark udf 内存利用率高

Question

我正在使用用 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: Lost task 40.0 in stage 4.0 (TID 183, ip-10-100-5-196.ec2.internal, executor 19): ExecutorLostFailure (executor 19 exited caused by one of the running tasks) Reason: Container killed by YARN for exceeding memory limits. 4.4 GB of 4.4 GB physical memory used. Consider boosting spark.yarn.executor.memoryOverhead.

最后它失败了。

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

Answer 1

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



当您使用 udf 时，由于您本地定义的函数未在 native JVM 结构中注册，因此无法通过简单的 java API 调用来实现，因此必须将其序列化/反序列化为 Python worker。然后数据将在 Python worker 中处理并序列化/反序列化回 JVM。

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

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