Spark“第一个”窗口功能比“最后一个”花费的时间长得多

Question

我正在使用 pyspark 例程来插入配置表中的缺失值。

想象一个从 0 到 50,000 的配置值表。用户在两者之间指定几个数据点（比如 0、50、100、500、2000、500000），然后我们对剩余部分进行插值。我的解决方案主要遵循 this blog post，但我没有使用任何 UDF。

在对此性能进行故障排除（大约需要 3 分钟）时，我发现一个特定的窗口函数一直在占用所有时间，而我正在做的其他事情只需要几秒钟。

这里是主要感兴趣的区域 - 我使用窗口函数来填写上一个和下一个用户提供的配置值：

from pyspark.sql import Window, functions as F

# Create partition windows that are required to generate new rows from the ones provided
win_last = Window.partitionBy('PORT_TYPE', 'loss_process').orderBy('rank').rowsBetween(Window.unboundedPreceding, 0)
win_next = Window.partitionBy('PORT_TYPE', 'loss_process').orderBy('rank').rowsBetween(0, Window.unboundedFollowing)

# Join back in the provided config table to populate the "known" scale factors
df_part1 = (df_scale_factors_template
  .join(df_users_config, ['PORT_TYPE', 'loss_process', 'rank'], 'leftouter')
  # Add computed columns that can lookup the prior config and next config for each missing value
  .withColumn('last_rank', F.last( F.col('rank'),         ignorenulls=True).over(win_last))
  .withColumn('last_sf',   F.last( F.col('scale_factor'), ignorenulls=True).over(win_last))
).cache()
debug_log_dataframe(df_part1 , 'df_part1') # Force a .count() and time Part1

df_part2 = (df_part1
  .withColumn('next_rank', F.first(F.col('rank'),         ignorenulls=True).over(win_next))
  .withColumn('next_sf',   F.first(F.col('scale_factor'), ignorenulls=True).over(win_next))
).cache()
debug_log_dataframe(df_part2 , 'df_part2') # Force a .count() and time Part2

df_part3 = (df_part2
  # Implements standard linear interpolation: y = y1 + ((y2-y1)/(x2-x1)) * (x-x1)
  .withColumn('scale_factor', 
              F.when(F.col('last_rank')==F.col('next_rank'), F.col('last_sf')) # Handle div/0 case
              .otherwise(F.col('last_sf') + ((F.col('next_sf')-F.col('last_sf'))/(F.col('next_rank')-F.col('last_rank'))) * (F.col('rank')-F.col('last_rank'))))
  .select('PORT_TYPE', 'loss_process', 'rank', 'scale_factor')
).cache()
debug_log_dataframe(df_part3, 'df_part3', explain: True) # Force a .count() and time Part3

上面曾经是一个单链数据框语句，但我已经将它分成 3 个部分，以便我可以隔离需要这么长时间的部分。结果是：

• Part 1: Generated 8 columns and 300006 rows in 0.65 seconds
• Part 2: Generated 10 columns and 300006 rows in 189.55 seconds
• Part 3: Generated 4 columns and 300006 rows in 0.24 seconds

---

为什么我对first() 的调用比Window.unboundedFollowing 比last() 比Window.unboundedPreceding 花费的时间长这么多？

---

一些避免问题/疑虑的注意事项：

• debug_log_dataframe 只是一个辅助函数，用于强制使用 .Count() 执行数据帧/缓存并计时以产生上述日志。
• 我们实际上是同时操作 6 个 50001 行的配置表（因此需要分区和行数）
• 作为一个健全的检查，我已经通过明确的unpersist()ing 在安排后续运行之前排除了cache() 重用的影响 - 我对上述测量结果非常有信心。

物理计划： 为了帮助回答这个问题，我在 part3 的结果上致电 explain() 以确认缓存正在产生预期的效果。此处注释以突出问题区域：

我能看到的唯一区别是：

• 前两个调用（对last）显示RunningWindowFunction，而对next 的调用只显示Window
• 第 1 部分旁边有一个 *(3)，但第 2 部分没有。

我尝试过的一些事情：

• 我尝试将第 2 部分进一步拆分为单独的数据帧 - 结果是每个 first 语句占用总时间的一半（约 98 秒）
• 我尝试颠倒生成这些列的顺序（例如，在调用 'first' 之后调用 'last'），但没有区别。无论哪个数据帧最终包含对first 的调用都是慢的。

我觉得我已经尽我所能地进行了挖掘，并且有点希望火花专家能够看一看知道这个时间来自哪里。

Answer 1

无法回答问题的解决方案

在尝试各种方法来加快我的日常工作时，我突然想到尝试将我对 first() 的用法重新改写为 last() 的用法，并采用相反的排序顺序。

所以重写这个：

win_next = (Window.partitionBy('PORT_TYPE', 'loss_process')
  .orderBy('rank').rowsBetween(0, Window.unboundedFollowing))

df_part2 = (df_part1
  .withColumn('next_rank', F.first(F.col('rank'),         ignorenulls=True).over(win_next))
  .withColumn('next_sf',   F.first(F.col('scale_factor'), ignorenulls=True).over(win_next))
)

这样：

win_next = (Window.partitionBy('PORT_TYPE', 'loss_process')
  .orderBy(F.desc('rank')).rowsBetween(Window.unboundedPreceding, 0))

df_part2 = (df_part1
  .withColumn('next_rank', F.last(F.col('rank'),         ignorenulls=True).over(win_next))
  .withColumn('next_sf',   F.last(F.col('scale_factor'), ignorenulls=True).over(win_next))
)

令我惊讶的是，这实际上解决了性能问题，现在整个数据帧仅在 3 秒内生成。我很高兴，但仍然很烦恼。

正如我预测的那样，查询计划现在在创建接下来的两列之前包含一个新的 SORT 步骤，并且它们已从 Window 更改为前两列的 RunningWindowFunction。这是新计划（不再将代码分解为 3 个单独的缓存部分，因为那只是为了解决性能问题）：

至于问题：

为什么我通过 Window.unboundedFollowing 调用 first() 比通过 Window.unboundedPreceding 调用 last() 花费这么多时间？

出于学术原因，我希望仍然有人能回答这个问题