Spark DataFrame で最も高い値を持つ各グループの最初の行を効率的に選択するにはどうすればよいですか?

各グループのホームページの選び方

目標は、DataFrame の各グループから最高値を持つ最初のページを抽出することです。これは、データの傾向を分析したり、特定のカテゴリの上位プレーヤーを特定したりするときによく当てはまります。これを実現するには、いくつかの方法を使用できます:

ウィンドウ関数:

ウィンドウ関数は、グループ内で計算を実行する方法を提供します。この場合、row_number() 関数を使用して、指定された順序に基づいて各行にシーケンス番号を割り当てることができます。ランキングを 1 にフィルタリングすることで、ホームページを識別できます。

<code class="language-scala">import org.apache.spark.sql.functions.{row_number, max, broadcast}
import org.apache.spark.sql.expressions.Window

val df = sc.parallelize(Seq(
  (0,"cat26",30.9), (0,"cat13",22.1), (0,"cat95",19.6), (0,"cat105",1.3),
  (1,"cat67",28.5), (1,"cat4",26.8), (1,"cat13",12.6), (1,"cat23",5.3),
  (2,"cat56",39.6), (2,"cat40",29.7), (2,"cat187",27.9), (2,"cat68",9.8),
  (3,"cat8",35.6))).toDF("Hour", "Category", "TotalValue")

val w = Window.partitionBy($"hour").orderBy($"TotalValue".desc)

val dfTop = df.withColumn("rn", row_number.over(w)).where($"rn" === 1).drop("rn")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+</code>

単純な SQL 集約後の結合:

あるいは、SQL を使用して集計を実行し、その結果を元の DataFrame と連結して各グループの最初のページを抽出することもできます。

<code class="language-scala">val dfMax = df.groupBy($"hour".as("max_hour")).agg(max($"TotalValue").as("max_value"))

val dfTopByJoin = df.join(broadcast(dfMax),
    ($"hour" === $"max_hour") && ($"TotalValue" === $"max_value"))
  .drop("max_hour")
  .drop("max_value")

dfTopByJoin.show

// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+</code>

構造ソートを使用します:

ウィンドウ関数や結合を使用せずに同じ結果を得る優れた方法は、値とカテゴリを含む構造に基づいてデータを並べ替えることです。この構造体の最大値は、各グループの必要な最初のページを返します。

<code class="language-scala">val dfTop = df.select($"Hour", struct($"TotalValue", $"Category").alias("vs"))
  .groupBy($"hour")
  .agg(max("vs").alias("vs"))
  .select($"Hour", $"vs.Category", $"vs.TotalValue")

dfTop.show
// +----+--------+----------+
// |Hour|Category|TotalValue|
// +----+--------+----------+
// |   0|   cat26|      30.9|
// |   1|   cat67|      28.5|
// |   2|   cat56|      39.6|
// |   3|    cat8|      35.6|
// +----+--------+----------+</code>

以上がSpark DataFrame で最も高い値を持つ各グループの最初の行を効率的に選択するにはどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。