hadoop - 在 Spark 上进行 rank() 的有效方法？

Question

我在 PySpark 上有一个三列数据框，我正在尝试在 SQL 上执行与 RANK() OVER(PARTITION BY ... ORDER BY ...) 等效的操作。数据框 df 看起来像:

col1, col2, score
A   , B   , 0.500
...

我知道我可以为此使用窗口函数:

from pyspark.sql.window import Window
from pyspark.sql import functions as F

windowSpec = Window.partitionBy(df['col1']).orderBy(df['score'].desc())
df = df.select('col1', 'col2', 'score', F.rank().over(windowSpec).alias('rnk'))

对于非常大的 df，这将需要花费大量时间跨分区打乱数据。有什么办法可以提高效率吗？ (如果不需要，我不需要使用数据框。)

Answer 1

添加到@DanieldePaula 的评论；在 Spark SQL 中，避免随机播放很棘手(/通常不可能)。我已经与 spark sql Michael Armbrust 的牧羊人讨论过这个问题 - 理由是如果允许使用 preservesPartioning=true 的等价物(来自许多spark sql 查询中的核心 RDD 操作。

在任何情况下，对于您的查询，如果不进行混洗就很难获得结果 - 因为您的数据甚至还没有正确预分区。

如果您希望获得更好的性能，您需要:

• 完全退出 spark sql
• 根据所需窗口对数据进行预分区
• 通过核心 RDD 原语手动执行窗口操作的等价物。

现在这些听起来有点严重繁重..所以..您可能只想享受窗口的性能。