在每组数据框 spark scala 中选择前 n 行答案

【问题标题】：selecting top n rows in each group of dataframe spark scala在每组数据框 spark scala 中选择前 n 行
【发布时间】：2019-12-29 22:17:40
【问题描述】：

我在 spark scala 中有一个 df，我需要根据第二列中的 DateTime 进行 groupBy id 和排序，并且只取每个组的前 5 行

------------------------------------
|id|             DateTime          |
------------------------------------
|340054675199675|15-01-2018 19:43:23|
|340054675199675|15-01-2018 10:56:43|
|340028465709212|10-01-2018 02:47:11|
|340054675199675|09-01-2018 10:59:10|
|340028465709212|02-01-2018 03:25:35|
|340054675199675|28-12-2017 05:48:04|
|340054675199675|21-12-2017 15:47:51|
|340028465709212|18-12-2017 10:33:04|
|340028465709212|16-12-2017 19:55:40|
|340028465709212|16-12-2017 19:55:40|
|340028465709212|12-12-2017 07:04:51|
|340054675199675|06-12-2017 08:52:38|
------------------------------------

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

    val dfMax = df.groupBy($"id".as("grouped_id")).agg(first($"DateTime").as("max_value")).limit(10)

    val dfTopByJoin = df.join(broadcast(dfMax),
      ($"id" === $"grouped_id") && ($"DateTime" === $"max_value"))

【问题讨论】：

您是否在使用 spark 流式传输 [给定标签]？
是的，我将从流媒体中获取这些数据
是 Dstreams 还是结构化流式传输？
结构化。但截至目前，我更感兴趣的是对上面的 df 进行排序。
为什么不这样做： val w= Window.partitionBy("id") val dfTop = df.withColumn("rn", row_number.over(w)).where($ "rn"

标签： scala apache-spark apache-spark-sql spark-streaming

【解决方案1】：

实现所需输出的 Scala 代码

import org.apache.spark.sql.expressions.Window
scala> df2.show
+---------------+-------------------+
|             id|           DateTime|
+---------------+-------------------+
|340054675199675|15-01-2018 19:43:23|
|340054675199675|15-01-2018 10:56:43|
|340028465709212|10-01-2018 02:47:11|
|340054675199675|09-01-2018 10:59:10|
|340028465709212|02-01-2018 03:25:35|
|340054675199675|28-12-2017 05:48:04|
|340054675199675|21-12-2017 15:47:51|
|340028465709212|18-12-2017 10:33:04|
|340028465709212|16-12-2017 19:55:40|
|340028465709212|16-12-2017 19:55:40|
|340028465709212|12-12-2017 07:04:51|
|340054675199675|06-12-2017 08:52:38|
+---------------+-------------------+


scala> df2.printSchema
root
   |-- id: string (nullable = true)
   |-- DateTime: string (nullable = true)

Dataframe COLUMN(DateTime) 是字符串格式，所以需要转换成时间戳，方便我们根据需求对数据进行排序。

 var df3 = df2.withColumn("DateTime",to_timestamp($"DateTime","dd-MM-yyyy HH:mm:ss")
 scala> df3.printSchema
 root
   |-- id: string (nullable = true)
   |-- DateTime: timestamp (nullable = true)

应用窗口函数来检索所需的输出

 val w= Window.partitionBy("id").orderBy("DateTime")
 val dfTop = df3.withColumn("rn", row_number.over(w)).filter($"rn"<6).drop(col("rn"))

 scala> dfTop.show
 +---------------+-------------------+
 |             id|           DateTime|
 +---------------+-------------------+
 |340028465709212|2017-12-12 07:04:51|
 |340028465709212|2017-12-16 19:55:40|
 |340028465709212|2017-12-16 19:55:40|
 |340028465709212|2017-12-18 10:33:04|
 |340028465709212|2018-01-02 03:25:35|
 |340054675199675|2017-12-06 08:52:38|
 |340054675199675|2017-12-21 15:47:51|
 |340054675199675|2017-12-28 05:48:04|
 |340054675199675|2018-01-09 10:59:10|
 |340054675199675|2018-01-15 10:56:43|
 +---------------+-------------------+

然后你会得到你想要的答案.HAppy HAdooooop

【讨论】：