加入性能问题：数据帧与应用了 UDF 的相同数据帧 [重复]答案

【问题标题】：Join performance issue: dataframe vs the same dataframe with a UDF applied [duplicate]加入性能问题：数据帧与应用了 UDF 的相同数据帧 [重复]
【发布时间】：2019-12-10 23:27:29
【问题描述】：

有两个数据框。一个 - df_table_a 是通过直接从 Hive 表中读取来创建的。另一个 - df_table_a_slow 是通过在 df_table_a 之上应用 UDF 转换创建的

df_table_a = spark.sql('SELECT INT_COL, DATE_WEEK FROM table_a')

shift_date_udf = F.udf(lambda day: day - pd.offsets.Week(3, weekday=0), DateType())
df_table_a_slow = df_fast.withColumn('DATE_WEEK', shift_date_udf('DATE_WEEK'))

然后是 df_table_b 数据框，它也是通过直接从 Hive 表中读取来创建的。

df_table_b = spark.sql('SELECT INT_COL, DATE_WEEK, OTHER_COLUMN FROM table_b')

现在我们将df_table_b 加入到上面定义的两个数据帧中。

df_fast_join = df_table_a.join(df, on=['INT_COL', 'DATE_WEEK'], how='left')
df_slow_join = df_table_a_slow.join(df, on=['INT_COL', 'DATE_WEEK'], how='left')

我想计算两个连接的执行时间，所以这里有一个函数来估计转换时间：

def time_transformation(df, subject):
    start = time.time()
    cnt = df.count()
    end = time.time()
    print(f'>>> {subject} - {cnt}', end - start)

结果：

time_transformation(df_fast_join, 'fast join')
 >> fast join - 75739267 37.43
time_transformation(df_slow_join, 'slow join')
 >>  slow join - 75739267 553.32

UDF 转换本身似乎不需要太多时间：

time_transformation(df_slow, 'df_slow')
  >> df_slow - 75739267 0.25

两个连接的执行计划相差一行：

+- BatchEvalPython [<lambda>(DATE_WEEK#1)], [INT_COL#0, DATE_WEEK#1, pythonUDF0#843]

问题：为什么将 UDF 应用于其中一个数据帧会使连接速度减慢 10 倍以上？怎么解决？

【问题讨论】：

udf 不是 spark 中的自然函数，因此它总是比通常的 spark 过程慢。即使您认为这是一项非常简单且微不足道的工作，但实际上还没有达到火花。
我怀疑 df_slow 实际上没有进行任何计算，这就是为什么它看起来很快，但尝试进行收集，这会慢很多

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

Python UDF 是如何工作的？

当在执行器内部处理一行时，它会被序列化，然后发送到 Python 解释器。然后在 Python 解释器中对该行进行反序列化，并在其上应用 UDF。此 UDF 将对行进行更改。现在 Row 将再次被序列化并发送回执行程序 JVM，执行程序 JVM 将其反序列化并做进一步的工作。所以这个额外的序列化-反序列化-序列化-反序列化会显着减慢每一行的处理速度，因此 python UDF 非常慢。

如何加快速度？

如果您可以使用一些 SQL 运算符或 Scala UDF 实现相同的效果，这将运行得更快。

【讨论】：

【解决方案2】：

UDF 函数很慢。特别是如果您使用 Python：数据帧本质上是 JVM 对象，而 Python UDF 一次应用一行。每次应用它们时，代码都会被序列化。看看here。从链接中您可以阅读：

这些用户定义的函数一次运行一行，因此遭受高序列化和调用开销。因此，许多数据管道在 Java 和 Scala 中定义 UDF，然后从 Python 调用它们。

【讨论】：