Spark 函数与 UDF 性能？

Question

Spark 现在提供了可在数据帧中使用的预定义函数，而且它们似乎已经过高度优化。我最初的问题是哪个更快，但我自己做了一些测试，发现火花函数至少在一个实例中快了大约 10 倍。有谁知道为什么会这样，udf 什么时候会更快（仅适用于存在相同 spark 函数的实例）？

这是我的测试代码（在 Databricks 社区版上运行）：

# UDF vs Spark function
from faker import Factory
from pyspark.sql.functions import lit, concat
fake = Factory.create()
fake.seed(4321)

# Each entry consists of last_name, first_name, ssn, job, and age (at least 1)
from pyspark.sql import Row
def fake_entry():
  name = fake.name().split()
  return (name[1], name[0], fake.ssn(), fake.job(), abs(2016 - fake.date_time().year) + 1)

# Create a helper function to call a function repeatedly
def repeat(times, func, *args, **kwargs):
    for _ in xrange(times):
        yield func(*args, **kwargs)
data = list(repeat(500000, fake_entry))
print len(data)
data[0]

dataDF = sqlContext.createDataFrame(data, ('last_name', 'first_name', 'ssn', 'occupation', 'age'))
dataDF.cache()

UDF 函数：

concat_s = udf(lambda s: s+ 's')
udfData = dataDF.select(concat_s(dataDF.first_name).alias('name'))
udfData.count()

火花功能：

spfData = dataDF.select(concat(dataDF.first_name, lit('s')).alias('name'))
spfData.count()

两次运行，udf 通常需要大约 1.1 - 1.4 s，而 Spark concat 函数总是需要不到 0.15 s。

Answer 1

在恢复使用您自己的自定义 UDF 函数之前，尽可能将更高级别的标准基于列的函数与数据集运算符一起使用，因为 UDF 是 Spark 的 BlackBox，因此它甚至不是尝试对其进行优化。

在屏幕后面实际发生的情况是，Catalyst 根本无法处理和优化 UDF，它会像 BlackBox 一样威胁它们，从而导致失去许多优化，如谓词下推、常量折叠和许多其他优化。

Answer 2

多年后，当我对火花知识有了更多了解并重新审视这个问题时，才意识到@alfredox 真正想问的是什么。于是我又修改了一遍，把答案分成两部分：

---

回答为什么原生DF函数（原生Spark-SQL函数）更快：

基本上，为什么原生 Spark 函数总是比 Spark UDF 快，无论您的 UDF 是用 Python 还是 Scala 实现的。

首先我们要了解Tungsten是什么，也就是firstly introduced in Spark 1.4。

它是一个后端，它的重点是什么：

1. 堆外内存管理使用二进制内存中数据表示（又名 Tungsten 行格式）并显式管理内存，
2. Cache Locality 是关于缓存感知计算和缓存感知布局以实现高缓存命中率，
3. 全阶段代码生成（又名 CodeGen）。

最大的 Spark 性能杀手之一是 GC。 GC 会暂停 JVM 中的每个线程，直到 GC 完成。这正是引入堆外内存管理的原因。

在执行 Spark-SQL 原生函数时，数据将保留在 tungsten 后端。但是，在 Spark UDF 场景中，数据将从 tungsten 移出到 JVM（Scala 场景）或 JVM 和 Python 进程（Python）进行实际处理，然后再移回 tungsten。结果：

1. 不可避免地会产生开销/惩罚：
   1. 反序列化来自 tungsten 的输入。
   2. 将输出序列化回钨。
2. 即使使用 Spark 中的一等公民 Scala，它也会增加 JVM 中的内存占用，并且可能会在 JVM 中涉及更多的 GC。 这个问题正是 tungsten 的“堆外内存管理”功能试图解决的问题。

---

回答 Python 是否一定比 Scala 慢：

自 2017 年 10 月 30 日起，Spark 刚刚为 pyspark 引入了矢量化 udf。

https://databricks.com/blog/2017/10/30/introducing-vectorized-udfs-for-pyspark.html

Python UDF 慢的原因，可能是 PySpark UDF 没有以最优化的方式实现：

根据链接中的段落。

Spark 在 0.7 版中添加了 Python API，支持用户定义的函数。这些用户定义的函数一次运行一行，因此受到高序列化和调用开销的影响。

不过，新的矢量化 udf 似乎大大提高了性能：

范围从 3 倍到超过 100 倍。

Answer 3

udf 什么时候会更快

如果您询问有关 Python UDF 的问题，答案可能永远不会*。由于 SQL 函数相对简单并且不是为复杂任务而设计的，因此几乎不可能补偿 Python 解释器和 JVM 之间重复序列化、反序列化和数据移动的成本。

有谁知道为什么会这样

主要原因已经在上面列举了，可以归结为一个简单的事实，Spark DataFrame 是原生的 JVM 结构，标准访问方法是通过对 Java API 的简单调用来实现的。另一方面，UDF 是用 Python 实现的，需要来回移动数据。

虽然 PySpark 通常需要在 JVM 和 Python 之间移动数据，但对于低级别的 RDD API，它通常不需要昂贵的 serde 活动。 Spark SQL 增加了序列化和序列化的额外成本，以及将数据从 JVM 上的不安全表示移动到不安全表示的成本。后者特定于所有 UDF（Python、Scala 和 Java），但前者特定于非本地语言。

与 UDF 不同，Spark SQL 函数直接在 JVM 上运行，并且通常与 Catalyst 和 Tungsten 很好地集成。这意味着这些可以在执行计划中进行优化，并且大多数时候可以从 codgen 和其他 Tungsten 优化中受益。此外，这些可以对其“本机”表示形式的数据进行操作。

所以从某种意义上说，这里的问题是 Python UDF 必须将数据带入代码，而 SQL 表达式则相反。

---

* 根据rough estimatesPySpark 窗口UDF 可以击败Scala 窗口函数。