performance - Spark 函数与 UDF 性能？

Question

Spark 现在提供了可以在数据帧中使用的预定义函数，而且它们似乎经过了高度优化。我最初的问题是关于哪个更快，但我自己做了一些测试，发现 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()

Spark 功能:

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

多次运行，udf 通常需要大约 1.1 - 1.4 秒，而 Spark concat函数总是在 0.15 秒以下。

Answer 1

when would a udf be faster

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

Does anyone know why this is so

上面已经列举了主要原因，可以归结为一个简单的事实，即 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 优化中受益。此外，这些可以在其“ native ”表示中对数据进行操作。

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

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