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()

火花功能:

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

什么时候会更快

如果您询问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表达式则相反.

*根据粗略估计 PySpark窗口UDF可以击败Scala窗口函数.

Answer 2

自2017年10月30日起,Spark刚刚为pyspark推出了矢量化udfs.

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

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

根据该链接的段落.

Spark在0.7版本中添加了一个Python API,支持用户定义的函数.这些用户定义的函数一次一行地运行,因此遭受高序列化和调用开销.

然而,新的矢量化udfs似乎正在大大提高性能:

从3倍到超过100倍.