각 그룹의 첫 번째 행은 어떻게 선택합니까?

source

각 그룹의 첫 번째 행은 어떻게 선택합니까?

manysource 2023. 7. 24. 22:36

각 그룹의 첫 번째 행은 어떻게 선택합니까?

다음과 같이 생성된 데이터 프레임이 있습니다.

df.groupBy($"Hour", $"Category")
  .agg(sum($"value") as "TotalValue")
  .sort($"Hour".asc, $"TotalValue".desc))

결과는 다음과 같습니다.

+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
|   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|
| ...|    ....|      ....|
+----+--------+----------+

보시다시피, 데이터 프레임은 다음과 같이 주문됩니다.Hour증가하는 순서대로, 그 다음에.TotalValue내림차순으로

각 그룹의 맨 위 행을 선택합니다.

Hour==0 그룹에서 (0,cat26,30.9)를 선택합니다.
Hour==1 그룹에서 (1,cat67,28.5)를 선택합니다.
Hour==2 그룹에서 (2,cat56,39.6)을 선택합니다.
등등

따라서 원하는 출력은 다음과 같습니다.

+----+--------+----------+
|Hour|Category|TotalValue|
+----+--------+----------+
|   0|   cat26|      30.9|
|   1|   cat67|      28.5|
|   2|   cat56|      39.6|
|   3|    cat8|      35.6|
| ...|     ...|       ...|
+----+--------+----------+

각 그룹의 상위 N개 행을 선택하는 것도 편리할 수 있습니다.

어떤 도움이든 대단히 감사합니다.

창 기능:

이와 같은 것이 효과가 있을 것입니다.

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|
// +----+--------+----------+

이 방법은 데이터 왜곡이 심한 경우 비효율적입니다.이 문제는 SPARK-34775에 의해 추적되며 나중에 해결될 수 있습니다(SPARK-37099).

일반 SQL 집계 뒤에 다음이 표시됩니다.

또는 다음과 같이 집계된 데이터 프레임에 가입할 수 있습니다.

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|
// +----+--------+----------+

총 값이 동일한 범주가 시간당 둘 이상인 경우 중복 값이 유지됩니다.다음과 같이 제거할 수 있습니다.

dfTopByJoin
  .groupBy($"hour")
  .agg(
    first("category").alias("category"),
    first("TotalValue").alias("TotalValue"))

다음을 통한 주문 사용:

잘 테스트되지는 않았지만 조인이나 창 기능이 필요 없는 깔끔한 트릭:

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|
// +----+--------+----------+

DataSet API(Spark 1.6+, 2.0+) 사용 시:

스파크 1.6:

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)
  .show

// +---+--------------+
// | _1|            _2|
// +---+--------------+
// |[0]|[0,cat26,30.9]|
// |[1]|[1,cat67,28.5]|
// |[2]|[2,cat56,39.6]|
// |[3]| [3,cat8,35.6]|
// +---+--------------+

스파크 2.0 이상:

df.as[Record]
  .groupByKey(_.Hour)
  .reduceGroups((x, y) => if (x.TotalValue > y.TotalValue) x else y)

마지막 두 가지 방법은 지도 측면 결합을 활용할 수 있으며 전체 셔플이 필요하지 않으므로 대부분의 시간이 창 기능 및 조인에 비해 더 나은 성능을 보일 것입니다.의 Structured Streaming에서도 사용할 수 있습니다.completed출력 모드

사용 안 함:

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

그것은 효과가 있는 것처럼 보일 수 있습니다(특히,local(SPARK-16207, 관련 JIRA 문제 연결에 대한 Tzach Zohar의 크레딧 및 SPARK-30335 참조).

동일한 참고 사항이 다음에 적용됩니다.

df.orderBy(...).dropDuplicates(...)

내부적으로 동등한 실행 계획을 사용합니다.

여러 열로 그룹화된 스파크 2.0.2의 경우:

import org.apache.spark.sql.functions.row_number
import org.apache.spark.sql.expressions.Window

val w = Window.partitionBy($"col1", $"col2", $"col3").orderBy($"timestamp".desc)

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

이것은 0323의 답변과 완전히 동일하지만 SQL 쿼리 방식입니다.

데이터 프레임이 생성되고 등록된다고 가정합니다.

df.createOrReplaceTempView("table")
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|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      |
//+----+--------+----------+

창 기능:

sqlContext.sql("select Hour, Category, TotalValue from (select *, row_number() OVER (PARTITION BY Hour ORDER BY TotalValue DESC) as rn  FROM table) tmp where rn = 1").show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

일반 SQL 집계 후 조인:

sqlContext.sql("select Hour, first(Category) as Category, first(TotalValue) as TotalValue from " +
  "(select Hour, Category, TotalValue from table tmp1 " +
  "join " +
  "(select Hour as max_hour, max(TotalValue) as max_value from table group by Hour) tmp2 " +
  "on " +
  "tmp1.Hour = tmp2.max_hour and tmp1.TotalValue = tmp2.max_value) tmp3 " +
  "group by tmp3.Hour")
  .show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

구조체 순서 지정 사용:

sqlContext.sql("select Hour, vs.Category, vs.TotalValue from (select Hour, max(struct(TotalValue, Category)) as vs from table group by Hour)").show(false)
//+----+--------+----------+
//|Hour|Category|TotalValue|
//+----+--------+----------+
//|1   |cat67   |28.5      |
//|3   |cat8    |35.6      |
//|2   |cat56   |39.6      |
//|0   |cat26   |30.9      |
//+----+--------+----------+

데이터 설정 방법과 하지 않음이 원래 답변과 동일합니다.

사용할 수 있습니다.max_by()기능은 스파크 3.0부터!

https://spark.apache.org/docs/3.0.0-preview/api/sql/index.html#max_by

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

// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("table")

// Using SQL
val result = spark.sql("select Hour, max_by(Category, TotalValue) AS Category, max(TotalValue) as TotalValue FROM table group by Hour order by Hour")

// or Using DataFrame API
val result = df.groupBy("Hour").
  agg(expr("max_by(Category, TotalValue)").as("Category"), max("TotalValue").as("TotalValue")).
  sort("Hour")

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

Apache DataFu로 쉽게 이 작업을 수행할 수 있습니다(안토닌의 답변과 구현이 유사함).

import datafu.spark.DataFrameOps._

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

df.dedupWithOrder($"Hour", $"TotalValue".desc).show

결과적으로

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

(예, 결과는 시간별로 정렬되지 않지만 중요한 경우 언제든지 나중에 수행할 수 있습니다.)

상위 N개 행을 차지하기 위한 API인 dedupTopN도 있습니다.그룹화당 행 수가 많을 경우 다른 API인 dedupWithCombiner도 사용할 수 있습니다.

(전체 공개 - 저는 DataFu 프로젝트의 일부입니다.)

데이터 프레임 API로 이를 수행하는 좋은 방법은 argmax 로직을 이렇게 사용하는 것입니다.

  val df = 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")

  df.groupBy($"Hour")
    .agg(max(struct($"TotalValue", $"Category")).as("argmax"))
    .select($"Hour", $"argmax.*").show

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

패턴은 키별로 그룹화됩니다 => 각 그룹에 대해 작업을 수행합니다(예: 축소 => 데이터 프레임으로 돌아가기).

이 경우 데이터 프레임 추상화가 조금 번거롭다고 생각하여 RDD 기능을 사용했습니다.

 val rdd: RDD[Row] = originalDf
  .rdd
  .groupBy(row => row.getAs[String]("grouping_row"))
  .map(iterableTuple => {
    iterableTuple._2.reduce(reduceFunction)
  })

val productDf = sqlContext.createDataFrame(rdd, originalDf.schema)

아래 솔루션은 하나의 groupBy만 수행하고 maxValue가 포함된 데이터 프레임의 행을 한 번에 추출합니다.추가 가입 또는 Windows가 필요하지 않습니다.

import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.DataFrame

//df is the dataframe with Day, Category, TotalValue

implicit val dfEnc = RowEncoder(df.schema)

val res: DataFrame = df
    .groupByKey{(r) => r.getInt(0)}
    .mapGroups[Row]{(day: Int, rows: Iterator[Row]) => i.maxBy{(r) => r.getDouble(2)}}

윈도우가 @zero323에 의해 제안된 대로 작동하는 동안 나는 다음과 같은 것을 발견했습니다.latest_per_key = groupby(*keys).agg(F.max(F.col('timestamp'))의 샘플을 에 사용합니다.data.join(latest_per_key, how='semileft', on= keys + 'timestamp')더 빨리 작동합니다.

여기서 당신은 이렇게 할 수 있습니다.

val data = df.groupBy("Hour").agg(first("Hour").as("_1"),first("Category").as("Category"),first("TotalValue").as("TotalValue")).drop("Hour")

data.withColumnRenamed("_1","Hour").show

언급URL : https://stackoverflow.com/questions/33878370/how-to-select-the-first-row-of-each-group

'source' 카테고리의 다른 글

저장 프로시저를 사용하여 다른 MariaDB 서버에서 Mariadb의 데이터베이스 개체에 액세스 (0)	2023.07.24
%f를 Python에서 strftime()과 함께 사용하여 마이크로초 가져오기 (0)	2023.07.24
누군가 SQL 쿼리를 저작권으로 보호할 수 있습니까? (0)	2023.07.24
쿼리를 사용하여 VIEW ddl 가져오기 (0)	2023.07.24
Android에서 날짜와 시간을 포맷하는 방법은 무엇입니까? (0)	2023.07.24

현재글각 그룹의 첫 번째 행은 어떻게 선택합니까?

각종 프로그래밍 정보를 다루는 블로그입니다.

JSON, sql-server, Python, reactjs, css, Java, MySQL, jQuery, C, php, MongoDB, ajax, Oracle, Excel, JavaScript, angularJS, Wordpress, mariadb, Git, spring-boot,

Today :
Yesterday :

manysource

각 그룹의 첫 번째 행은 어떻게 선택합니까?

각 그룹의 첫 번째 행은 어떻게 선택합니까?

'source' 카테고리의 다른 글

'source'의 다른글

티스토리툴바

« 2024/12 »
일	월	화	수	목	금	토
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30	31

각 그룹의 첫 번째 행은 어떻게 선택합니까?

각 그룹의 첫 번째 행은 어떻게 선택합니까?

'source' 카테고리의 다른 글

'source'의 다른글

관련글

티스토리툴바