使用 Spark 按一列的连续相同值进行分组，每组取另一列的最大值或最小值答案

【问题标题】：Use Spark to group by consecutive same values of one column, taking Max or Min value of another column for each group使用 Spark 按一列的连续相同值进行分组，每组取另一列的最大值或最小值
【发布时间】：2020-10-22 00:35:47
【问题描述】：

假设我有以下数据框

+-------------------+------+------------+
|               Date|   Val|   Condition|
+-------------------+------+------------+
|2020-10-02 10:00:00|211.39|         Max|
|2020-10-02 10:10:00|210.94|         Min|
|2020-10-02 10:30:00|209.21|         Max|
|2020-10-02 11:20:00|207.48|         Min|
|2020-10-02 11:50:00|207.22|         Min| <- take only this row because it's less than 207.48
|2020-10-02 12:10:00|207.58|         Max|
|2020-10-02 12:40:00|207.45|         Min|
|2020-10-02 13:10:00|207.45|         Min| <- take either row becase they are equal
|2020-10-02 13:40:00| 208.7|         Max| <- take only this row because it's greater than 208.31
|2020-10-02 14:10:00|208.31|         Max| 
|2020-10-02 14:20:00|208.16|         Min|
|2020-10-02 14:30:00| 208.3|         Max|
|2020-10-02 14:50:00|208.25|         Min|
|2020-10-02 15:10:00| 208.7|         Max|
|2020-10-02 15:30:00|208.08|         Min|
|2020-10-02 16:00:00| 208.0|         Min| <- take only this row because it's less than 208.08
|2020-10-02 16:30:00|208.35|         Max|
|2020-10-02 16:40:00|208.26|         Min|
|2020-10-02 16:50:00|208.27|         Max|
|2020-10-02 17:30:00|208.06|         Min|
+-------------------+------+------------+

如何按Condition 的连续值对其进行分组，并为每个组取Val 的最大值或最小值？（例如，生成的数据框应该类似于下面的数据框）（见上述数据框中的 cmets）。

+-------------------+------+------------+
|               Date|   Val|   Condition|
+-------------------+------+------------+
|2020-10-02 10:00:00|211.39|         Max|
|2020-10-02 10:10:00|210.94|         Min|
|2020-10-02 10:30:00|209.21|         Max|
|2020-10-02 11:50:00|207.22|         Min|
|2020-10-02 12:10:00|207.58|         Max|
|2020-10-02 12:40:00|207.45|         Min|
|2020-10-02 13:40:00| 208.7|         Max|
|2020-10-02 14:20:00|208.16|         Min|
|2020-10-02 14:30:00| 208.3|         Max|
|2020-10-02 14:50:00|208.25|         Min|
|2020-10-02 15:10:00| 208.7|         Max|
|2020-10-02 16:00:00| 208.0|         Min|
|2020-10-02 16:30:00|208.35|         Max|
|2020-10-02 16:40:00|208.26|         Min|
|2020-10-02 16:50:00|208.27|         Max|
|2020-10-02 17:30:00|208.06|         Min|
+-------------------+------+------------+

目标是：

对于有多个连续行且 Condition = Max 或 Condition = Min 的每个组
从每组中只取一行（哪一行 - 由 Condition 的值决定 - 它是具有 Val 列最大值或最小值的行）

【问题讨论】：

你可以尝试按row_number除以2（四舍五入）分组
@madprogrammer 有点与您的输出 DF 混淆。你能详细说明一下吗？
@SathiyanS 我编辑了这个问题，试图让它不那么混乱
问题是它不能被并行化，尽管Window.orderBy("Date") 可能会给出预期的结果，但这只会使用 1 个核心并且不会随着更大的数据进行扩展。 @madprogrammer

标签： apache-spark

【解决方案1】：

试试这个，

val wind = Window.orderBy("Date")
    val df1 = df.withColumn("val1", when($"Condition" === lead($"Condition", 1).over(wind),
      when($"Condition" === "Min", min($"val").over(wind.rowsBetween(0,1))).otherwise(max($"val").over(wind.rowsBetween(0,1))))
        .when($"Condition" === lag($"Condition", 1).over(wind),
          when($"Condition" === "Min", min($"val").over(wind.rowsBetween(-1,0))).otherwise(max($"val").over(wind.rowsBetween(-1,0))))
      .otherwise($"val"))

    val df2 = df1.withColumn("rn", when($"Condition" === lead($"Condition", 1).over(wind),1)
      .when($"Condition" === lag($"Condition", 1).over(wind), 2)
      .otherwise(1)).withColumn("Val", $"val1").filter($"rn" === 1).drop("rn", "val1")

    df2.show(false)

+-------------------+------+---------+
|Date               |Val   |Condition|
+-------------------+------+---------+
|2020-10-02 10:00:00|211.39|Max      |
|2020-10-02 10:10:00|210.94|Min      |
|2020-10-02 10:30:00|209.21|Max      |
|2020-10-02 11:20:00|207.22|Min      |
|2020-10-02 12:10:00|207.58|Max      |
|2020-10-02 12:40:00|207.45|Min      |
|2020-10-02 13:40:00|208.7 |Max      |
|2020-10-02 14:20:00|208.16|Min      |
|2020-10-02 14:30:00|208.3 |Max      |
|2020-10-02 14:50:00|208.25|Min      |
|2020-10-02 15:10:00|208.7 |Max      |
|2020-10-02 15:30:00|208.0 |Min      |
|2020-10-02 16:30:00|208.35|Max      |
|2020-10-02 16:40:00|208.26|Min      |
|2020-10-02 16:50:00|208.27|Max      |
|2020-10-02 17:30:00|208.06|Min      |
+-------------------+------+---------+

如果对你有帮助，请告诉我。

【讨论】：

谢谢，很有帮助。同时，我设法获得了另一个似乎也有效的解决方案，请参阅我自己的答案。由于我对 Spark 有点陌生，如果你能评论它会很好，以防你发现任何问题。
我发现您和我的解决方案都选择了错误的日期（例如，对于 Val 207.22 的行，它应该是 2020-10-02 11:50:00 而不是 2020-10-02 11： 20:00)

【解决方案2】：

这个问题可以通过为连续行中具有相同条件的行准备额外的带有组号的列来解决（下面查询中的“组”列）。

val numOfPartitions = <set numer of partitions>
val window = Window.orderBy("Date")

df
  .withColumn("condition_change", when(col("Condition") === lag("Condition", 1, false).over(window), 0).otherwise(1))
  .withColumn("group", sum("condition_change").over(window))
  .drop("condition_change")
  .repartition(numOfPartitions)
  .groupBy("group")
  .agg(
    min(struct("Val", "Date")) as "min",
    max(struct("Val", "Date")) as "max",
    first("Condition") as "Condition")
  .withColumn("result", when(col("Condition") === "Min", col("min")).otherwise(col("max")))
  .select(col("result.Date") as "Date", col("result.Val") as "Val", col("Condition"))
  .show()

请注意，您必须设置numOfPartitions进行重新分区（否则任务将在一个执行器上运行），选择与您拥有的数据量匹配的值，首先尝试可以是“spark.sql.shuffle.partitions”的值。

【讨论】：

这似乎不准确。如果对于相同数量的连续行，条件保持不变，对于两种不同的情况，组变量将错误地将相同的组号分配给这些。

【解决方案3】：

我想出了以下解决方案，它绝不是优化的，可能可以改进，但似乎给出了正确的结果

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

val w1 = Window.orderBy("Date")
val w2 = Window.orderBy("Date")
  .rowsBetween(Window.unboundedPreceding, Window.currentRow)
val w3 = Window.partitionBy("cumsum").orderBy(lit(1))

val grpExtrema = extrema
.withColumn("dupe", when(lag("Condition", 1).over(w1) !== extrema("Condition"), 1).otherwise(0))
.withColumn("cumsum", sum("dupe").over(w2))
.drop("dupe")

grpExtrema
.withColumn("row", row_number.over(w3))
.withColumn("Val",
            when($"Condition" === lit("min"), min("Val").over(w3))
            .otherwise(max("Val").over(w3)))
.where(col("row") === 1).select("Date", "Val", "Condition")
.show()

【讨论】：