Spark で各グループの最上位行を効率的に選択するにはどうすればよいですか?

各グループの最初の行を効率的に選択する

この記事は、「時間」と「カテゴリ」のグループごとに「合計値」が最も高いデータ行を抽出することを目的としています。 これを行うにはいくつかの方法があります:

ウィンドウ関数を使用する:

ウィンドウ関数は、各グループ内で計算を実行する効率的な方法を提供します。 その方法の 1 つを次に示します:

<code>import org.apache.spark.sql.functions.{row_number, max, broadcast}
import org.apache.spark.sql.expressions.Window

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

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

SQL 集計と結合の使用:

もう 1 つのアプローチは、SQL 集約とその後の結合を利用することです。

<code>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")</code>

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

賢い方法は、「合計値」と「カテゴリ」を含む構造体をソートすることです:

<code>val dfTop = df.select($"Hour", struct($"TotalValue", $"Category").alias("vs"))
  .groupBy($"Hour")
  .agg(max("vs").alias("vs"))
  .select($"Hour", $"vs.Category", $"vs.TotalValue")</code>

DataSet API の使用 (Spark 1.6):

DataSet API は、同じ結果を達成するための簡潔な方法を提供します。

<code>case class Record(Hour: Integer, Category: String, TotalValue: Double)

df.as[Record]
  .groupBy($"Hour")
  .reduce((x, y) => if (x.TotalValue > y.TotalValue) x else y)</code>

間違いを避ける方法:

次の方法は信頼性の低い結果を生成する可能性があるため、避ける必要があります:

• df.orderBy(...).groupBy(...).agg(first(...), ...)
• df.orderBy(...).dropDuplicates(...)

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