Spark DataFrame에서 각 그룹의 첫 번째 행을 선택하는 방법은 무엇입니까?

각 그룹의 첫 번째 행 선택

특정 정렬 기준에 따라 각 그룹의 첫 번째 행을 검색하려면 다음과 같은 여러 방법을 사용할 수 있습니다.

창 기능

<code class="language-scala">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))

// 过滤排名为1的行
dfTop.where($"rn" === 1).drop("rn")</code>

일반 SQL 집계 및 조인

<code class="language-scala">// 聚合以查找每个小时的最大值
val dfMax = df.groupBy($"Hour".as("max_hour")).agg(max($"TotalValue").as("max_value"))

// 将原始DataFrame与聚合后的DataFrame连接
val dfTopByJoin = df.join(broadcast(dfMax), ($"Hour" === $"max_hour") && ($"TotalValue" === $"max_value"))

// 删除不必要的列
dfTopByJoin.drop("max_hour").drop("max_value")</code>

구조정렬

<code class="language-scala">// 为包含TotalValue和Category的结构体定义别名
val vs = struct($"TotalValue", $"Category").alias("vs")

// 按Hour分组并查找每个分组的最大结构体
val dfTop = df.select($"Hour", vs).groupBy($"Hour").agg(max(vs).alias("vs"))

// 从最大结构体中提取Category和TotalValue
dfTop.select($"Hour", $"vs.Category", $"vs.TotalValue")</code>

DataFrame API 사용

<code class="language-scala">// 为DataFrame定义一个自定义类
case class Record(Hour: Integer, Category: String, TotalValue: Double)

// 将DataFrame转换为自定义类
val dfRecords = df.as[Record]

// 按Hour分组并减少以查找TotalValue最大的记录
val dfTopRecords = dfRecords.groupByKey(_.Hour).reduce((x, y) => if (x.TotalValue > y.TotalValue) x else y)

// 转换回DataFrame
dfTopRecords.toDF</code>

위 내용은 Spark DataFrame에서 각 그룹의 첫 번째 행을 선택하는 방법은 무엇입니까?의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!