如何从以前的团队中获得价值？

Question

我需要在星星之火中获得前一组的值，并将其设置为当前组。我怎样才能做到这一点？我必须点菜，而不是TEXT_NUM。

由TEXT_NUM排序是不可能的，因为事件在时间上重复，如计数10和11所示。

我正在尝试使用以下代码：

   val spark = SparkSession.builder()
      .master("spark://spark-master:7077")
      .getOrCreate()

    val df = spark
      .createDataFrame(
        Seq[(Int, String, Int)](
          (0, "", 0),
          (1, "", 0),
          (2, "A", 1),
          (3, "A", 1),
          (4, "A", 1),
          (5, "B", 2),
          (6, "B", 2),
          (7, "B", 2),
          (8, "C", 3),
          (9, "C", 3),
          (10, "A", 1),
          (11, "A", 1)
        ))
      .toDF("count", "TEXT", "TEXT_NUM")

    val w1 = Window
      .orderBy("count")
      .rangeBetween(Window.unboundedPreceding, -1)
    df
      .withColumn("LAST_VALUE", last("TEXT_NUM").over(w1))
      .orderBy("count")
      .show()

结果：

+-----+----+--------+----------+
|count|TEXT|TEXT_NUM|LAST_VALUE|
+-----+----+--------+----------+
|    0|    |       0|      null|
|    1|    |       0|         0|
|    2|   A|       1|         0|
|    3|   A|       1|         1|
|    4|   A|       1|         1|
|    5|   B|       2|         1|
|    6|   B|       2|         2|
|    7|   B|       2|         2|
|    8|   C|       3|         2|
|    9|   C|       3|         3|
|   10|   A|       1|         3|
|   11|   A|       1|         1|
+-----+----+--------+----------+

预期结果：

+-----+----+--------+----------+
|count|TEXT|TEXT_NUM|LAST_VALUE|
+-----+----+--------+----------+
|    0|    |       0|      null|
|    1|    |       0|      null|
|    2|   A|       1|         0|
|    3|   A|       1|         0|
|    4|   A|       1|         0|
|    5|   B|       2|         1|
|    6|   B|       2|         1|
|    7|   B|       2|         1|
|    8|   C|       3|         2|
|    9|   C|       3|         2|
|   10|   A|       1|         3|
|   11|   A|       1|         3|
+-----+----+--------+----------+

Answer 1

考虑使用Window函数last(columnName, ignoreNulls)在组边界上由以前的"text_num“组成的列中回填null，如下所示：

val df = Seq(
  (0, "", 0), (1, "", 0),
  (2, "A", 1), (3, "A", 1), (4, "A", 1),
  (5, "B", 2), (6, "B", 2), (7, "B", 2),
  (8, "C", 3), (9, "C", 3),
  (10, "A", 1), (11, "A", 1)
).toDF("count", "text", "text_num")

import org.apache.spark.sql.expressions.Window
val w1 = Window.orderBy("count")
val w2 = w1.rowsBetween(Window.unboundedPreceding, 0)

df.
  withColumn("prev_num", lag("text_num", 1).over(w1)).
  withColumn("last_change", when($"text_num" =!= $"prev_num", $"prev_num")).
  withColumn("last_value", last("last_change", ignoreNulls=true).over(w2)).
  show
/*
+-----+----+--------+--------+-----------+----------+
|count|text|text_num|prev_num|last_change|last_value|
+-----+----+--------+--------+-----------+----------+
|    0|    |       0|    null|       null|      null|
|    1|    |       0|       0|       null|      null|
|    2|   A|       1|       0|          0|         0|
|    3|   A|       1|       1|       null|         0|
|    4|   A|       1|       1|       null|         0|
|    5|   B|       2|       1|          1|         1|
|    6|   B|       2|       2|       null|         1|
|    7|   B|       2|       2|       null|         1|
|    8|   C|       3|       2|          2|         2|
|    9|   C|       3|       3|       null|         2|
|   10|   A|       1|       3|          3|         3|
|   11|   A|       1|       1|       null|         3|
+-----+----+--------+--------+-----------+----------+
*/

中间列保存在输出中以供参考。如果不需要的话就扔掉吧。