如何在Spark中有效率地選擇每個組的頂行？

高效率選擇每個分組的首行

本文旨在擷取每個「小時」和「類別」分組中「總值」最高的那一行資料。 有多種方法可以實現：

使用視窗函數:

視窗函數提供了一種在每個分組內進行計算的有效方法。 以下是一種實作方式：

<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聚合和連接:

另一種方法是利用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中文網其他相關文章！