python之Spark 的 reduceByKey 最佳实践_编程设计

我有一个带有下一个模式的数据框:

root 
 |-- id_1: long (nullable = true) 
 |-- id_2: long (nullable = true) 
 |-- score: double (nullable = true)

数据看起来像:

+----+----+------------------+ 
|id_1|id_2|score             | 
+----+----+------------------+ 
|0   |9   |0.5888888888888889| 
|0   |1   |0.6166666666666667| 
|0   |2   |0.496996996996997 | 
|1   |9   |0.6222222222222221| 
|1   |6   |0.9082996632996633| 
|1   |5   |0.5927450980392157| 
|2   |3   |0.665774107440774 | 
|3   |8   |0.6872367465504721| 
|3   |8   |0.6872367465504721| 
|5   |6   |0.5365909090909091| 
+----+----+------------------+

目标是为每个得分最高的 id_1 找到 id_2。也许我错了，但是...只需要创建成对的 RDD:

root 
 |-- _1: long (nullable = true) 
 |-- _2: struct (nullable = true) 
 |    |-- _1: long (nullable = true) 
 |    |-- _2: double (nullable = true) 
 
+---+----------------------+ 
|_1 |_2                    | 
+---+----------------------+ 
|0  |[9,0.5888888888888889]| 
|0  |[1,0.6166666666666667]| 
|0  |[2,0.496996996996997] | 
|1  |[9,0.6222222222222221]| 
|1  |[6,0.9082996632996633]| 
|1  |[5,0.5927450980392157]| 
|2  |[3,0.665774107440774] | 
|3  |[8,0.6872367465504721]| 
|3  |[8,0.6872367465504721]| 
|5  |[6,0.5365909090909091]| 
+---+----------------------+

并用 max 键减少。有点像

paired_rdd.reduceByKey(lambda x1, x2: max(x1, x2, key=lambda x: x[-1]))

或者与 DataFrame API 相同(没有成对的 rdd):

original_df.groupBy('id_1').max('score')

我有两个问题，如果有人能指出我的错误步骤，我将不胜感激。

对于 10 亿甚至 1000 亿条记录:实现目标的最佳做法是什么(为每个 id_1 找到最高分的 id_2)？我已经尝试了 5000 万和 100M 的记录，并且在改组数据时取得了更好的结果(这与 Holden Karau 所说的相反)。我已按 id_1
完成重新分区
.repartition(X, "id_1")

然后 reduceByKey 速度更快。为什么？
为什么 DataFrame API 比 RDD API 慢几倍？我哪里错了？

谢谢。

请您参考如下方法:

您的用例是窗口聚合函数的完美用例。试一试，看看它与 RDD 的 reduceByKey 相比如何。

有时不是关于基于 RDD 的管道是否比基于 DataFrame 的管道更快，而是关于一个与另一个相比的表现力如何。基于 DataFrame 的管道几乎总是比基于 RDD 的替代方案更具表现力(从长远来看可能更易于维护)。

(我正在使用 Scala，将代码转换为 Python 作为家庭练习)

scala> dataset.show 
+----+----+------------------+ 
|id_1|id_2|             score| 
+----+----+------------------+ 
|   0|   9|0.5888888888888889| 
|   0|   1|0.6166666666666667| 
|   0|   2| 0.496996996996997| 
|   1|   9|0.6222222222222221| 
|   1|   6|0.9082996632996633| 
|   1|   5|0.5927450980392157| 
|   2|   3| 0.665774107440774| 
|   3|   8|0.6872367465504721| 
|   3|   8|0.6872367465504721| 
|   5|   6|0.5365909090909091| 
+----+----+------------------+ 
 
import org.apache.spark.sql.expressions.Window 
val byId_1 = Window.partitionBy("id_1") 
original_df.select($"id_1", max() over byId_1) 
scala> dataset. 
  select($"id_1", $"id_2", $"score", max("score") over byId_1 as "max_score"). 
  filter($"score" === $"max_score"). 
  distinct.  // <-- id_1 == 3 is duplicated 
  sort("id_1"). 
  show 
+----+----+------------------+------------------+ 
|id_1|id_2|             score|         max_score| 
+----+----+------------------+------------------+ 
|   0|   1|0.6166666666666667|0.6166666666666667| 
|   1|   6|0.9082996632996633|0.9082996632996633| 
|   2|   3| 0.665774107440774| 0.665774107440774| 
|   3|   8|0.6872367465504721|0.6872367465504721| 
|   5|   6|0.5365909090909091|0.5365909090909091| 
+----+----+------------------+------------------+

请注意，默认情况下，DataFrames 使用 spark.sql.shuffle.partitions，即 200，上周我遇到了一个案例，其中大部分分区(以及任务) 其中空的导致数以千计的任务等待执行，这是徒劳的，并且消耗了 CPU 周期。我们从几小时缩短到几秒钟。

了解数据及其分区方式是优化 Spark 查询的第一步，无论是使用 RDD API 还是 Dataset API 编写。