根据值对数据集行进行分组答案

【问题标题】：group the dataset rows based on the value根据值对数据集行进行分组
【发布时间】：2021-07-24 17:30:06
【问题描述】：

我需要基于某些标准的数据集的组行。我的输入数据集就像 df1 ：

+------------+----------+----------------------------+------------------+-------+
|     col_1  |     col_2|col_3                       |            tp    |  range|
+------------+----------+----------------------------+------------------+-------+
|MP          |W         |                           X|10                |]0,3]  |
|MP          |W         |                           X|20                |]12,30]|
|MP          |W         |                           X|18                |]12,30]|
|MP          |W         |                           X|18                |]0,3]  |
|MP          |W         |                           X|30                |]0,3]  |
|MP          |W         |                           X|50                |]12,30]|
|MP          |W         |                           X|18                |]12,30]|
|MP          |W         |                           X|60                |]12,30]|
|MP          |W         |                           X|50                |]12,30]|
|MP          |W         |                           X|70                |]12,30]|
|MP          |W         |                           X|18                |]12,30]|
|MP          |W         |                           X|90                |]12,30]|
|MP          |W         |                           X|18                |]36,48]|
|MP          |W         |                           X|18                |]36,48]|
|MP          |W         |                           X|18                |]12,30]|
|MP          |W         |                           X|180               |]12,30]|
|MP          |W         |                           X|18                |]36,48]|
|MP          |W         |                           X|18                |]12,30]|
|MP          |W         |                         S2E|19                |]24,36]|
|MP          |W         |                         S2E|40                |]24,36]|
+------------+----------+----------------------------+------------------+-------+

我想做的是：

按范围分组 df1 的行（最后一列）[df = df1.select("*").groupby("col_1", "col_2", "col_3", "tp", "range"]
对于同一范围内的行，创建子组，其中同一子组的 2 个产量（列名 = tp）之间的比率小于 2 [即 tp(i-1)/tp(i)

在 ]12,30] 范围的输出中，我将有类似的内容：

+------------+----------+----------------------------+------------------+-------+------------+
|     col_1  |     col_2|col_3                       |            tp    |  range|  subgroup  |
+------------+----------+----------------------------+------------------+-------+------------+
|MP          |W         |                           X|20                |]12,30]|subgroup_1  |
|MP          |W         |                           X|18                |]12,30]|subgroup_1  |
|MP          |W         |                           X|50                |]12,30]|subgroup_2  |
|MP          |W         |                           X|18                |]12,30]|subgroup_1  |
|MP          |W         |                           X|60                |]12,30]|subgroup_2  |
|MP          |W         |                           X|50                |]12,30]|subgroup_2  |
|MP          |W         |                           X|70                |]12,30]|subgroup_2  |
|MP          |W         |                           X|90                |]12,30]|subgroup_2  |
|MP          |W         |                           X|180               |]12,30]|subgroup_3  |
+------------+----------+----------------------------+------------------+-------+------------+

有人有解决方案吗？我正在使用 Spark Java。

【问题讨论】：

tp(i-1)/tp(i) < 2 or tp(i-2)/tp(i) < 2 到底是什么意思？就像用前一行扫描每一行并将结果设置为subgroup 列中的标签一样？什么是tp(i)、tp(i-1)。 tp(i-2) 在这里，基于您的（分组的）DataFrame？
With (tp(i-1)/tp(i) 我的意思是，在创建子组之后，如果我在该子组中随机取 2 行，则它们的 tp 值（数据集的列“tp”）之间的比率必须小于 2。可能是输出数据集更好地说明。感谢您的帮助。
不是machine-learning 问题，请不要向无关标签发送垃圾邮件（已删除）。

标签： java apache-spark apache-spark-sql apache-spark-dataset

【解决方案1】：

首先，col_1、col_2、col_3 和 range 列无关紧要。它们可以通过group 列抽象出来。

想法是使用窗口函数按tp值对每个窗口中的行进行排序，然后：

为每一行创建一个行号，稍后将用作子组 ID。
计算每一行与其前一行的比率
如果比率大于等于2，则使用当前行的行号作为子组id；否则，从上一行结转子组 ID。

scala 中的代码，但应该展示这个想法：

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

val data = Seq(
  (1, 20),
  (1, 18),
  (1, 50),
  (1, 18),
  (1, 60),
  (1, 50),
  (1, 70),
  (1, 90),
  (1, 180),
  (1, 360)
) toDF ("group", "tp")

val windowSpec = Window.partitionBy($"group").orderBy($"tp")
val df = data
  .withColumn("lag_tp", lag($"tp", 1, 0).over(windowSpec))
  .withColumn("row_num", row_number.over(windowSpec))
  .withColumn("reci_yield", $"lag_tp" / $"tp")
  .withColumn("yield_ge_2", $"reci_yield" <= 0.5)
  .withColumn("subGroup", 
                // When yield >= 2 detected, get the current row number as subGroup id
                when($"yield_ge_2" === true, $"row_num") 
                .otherwise(
                  // otherwise, get the last non-null subGroup id.
                   last(
                     when($"yield_ge_2"===true, $"row_num"), 
                     ignoreNulls = true
                   ).over(windowSpec)
                )
             )
  // drop intermediate columns
  .drop("row_num", "lag_tp", "reci_yield", "yield_ge_2")

df.show(false)

输出：

+-----+---+--------+
|group|tp |subGroup|
+-----+---+--------+
|1    |18 |1       |
|1    |18 |1       |
|1    |20 |1       |
|1    |50 |4       |
|1    |50 |4       |
|1    |60 |4       |
|1    |70 |4       |
|1    |90 |4       |
|1    |180|9       |
|1    |360|10      |
+-----+---+--------+

信用：https://stackoverflow.com/a/65373636/3546203

【讨论】：