如何使用 UDF 对 Spark SQL 结果进行后处理答案

【问题标题】：How to post-process Spark SQL results w/o using UDF如何使用 UDF 对 Spark SQL 结果进行后处理
【发布时间】：2019-03-10 17:18:34
【问题描述】：

我读过

建议不要使用UDF来节省反序列化/序列化成本。

就我而言，我做了这样的查询

select MYFUN(f1, f2, ...) 
from A ...

我使用MYFUN对查询结果逐行进行后处理，例如发送到另一个服务。

def my_fun(f1, f2, ...):
   service.send(f1, f2, ...)

session.udf.register('MYFUN', my_fun)

不使用 UDF，我可能想将查询结果保存到 Python 数据帧，或 hdfs 上的 Parque 表，然后按数据帧读取，并逐个处理数据帧。

问题是结果表很大，可能是1M行。这种情况下，去掉UDF还有意义吗？

将 Spark SQL 结果填充到另一个服务的最佳做法是什么？

【问题讨论】：

【解决方案1】：

从性能的角度来看，不建议使用 Python UDF，但在需要时使用它们并没有什么问题，就像在这种情况下：与您的 @ 引入的 I/O 等待相比，序列化/反序列化成本可能是荒谬的987654322@。所以删除 UDF 可能没有意义。

在更一般的情况下，有两种方法可以减少处理数据帧的内存占用。您已经提到的一种方法是保存到文件并处理文件。

另一种方法是在您的数据框上使用toLocalIterator。这样，您将迭代每个数据帧的分区：您可以重新分区数据帧以创建任意大小的分区：

df =df.repartition(100)
for partition in df.toLocalIterator():
    for row in partition:
        send(row)

这样，您的本地内存需求将减少到重新分区数据帧的最大分区。

【讨论】：