可以使用row_number函数将唯一标识符分配给每行，并通过Window使用范围找到前一个ID行。接收到的连接到自身的结构：

val df = Seq(
  ("2023-11-01T12:51", "ID", "75B"),
  ("2023-11-01T12:52", "Pressure", "5"),
  ("2023-11-01T12:56", "Resistance", "20"),
  ("2023-11-01T12:57", "ID", "55C"),
  ("2023-11-01T12:57", "Pressure", "10")
).toDF("Time", "Name", "Value")

val withCurrentRowWindow = Window.orderBy("Time")
  .rangeBetween(Window.unboundedPreceding, Window.currentRow)

val withIdRowDetected = df
  .withColumn("row_number", row_number().over(Window.orderBy("Time")))

  .withColumn("id_row_number", max(
    when($"Name" === lit("ID"), $"row_number").otherwise(null)
  ).over(withCurrentRowWindow))

withIdRowDetected.show(false)

val result = withIdRowDetected
  .alias("non_ids")
  .where($"Name" =!= lit("ID"))
  .join(withIdRowDetected
    .alias("ids")
    .where($"Name" === lit("ID")),
    $"non_ids.id_row_number" === $"ids.row_number", "inner")
  .select($"non_ids.Time", $"non_ids.Name", $"non_ids.Value", $"ids.Value".alias("ID"))

字符串
输出量：

+----------------+----------+-----+----------+-------------+
|Time            |Name      |Value|row_number|id_row_number|
+----------------+----------+-----+----------+-------------+
|2023-11-01T12:51|ID        |75B  |1         |1            |
|2023-11-01T12:52|Pressure  |5    |2         |1            |
|2023-11-01T12:56|Resistance|20   |3         |1            |
|2023-11-01T12:57|ID        |55C  |4         |4            |
|2023-11-01T12:57|Pressure  |10   |5         |4            |
+----------------+----------+-----+----------+-------------+

+----------------+----------+-----+---+
|Time            |Name      |Value|ID |
+----------------+----------+-----+---+
|2023-11-01T12:52|Pressure  |5    |75B|
|2023-11-01T12:56|Resistance|20   |75B|
|2023-11-01T12:57|Pressure  |10   |55C|
+----------------+----------+-----+---+

型
注意：使用没有分区的窗口，性能可能会很差。

下面是我使用窗口函数的解决方案：

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

val data = Seq((5, "Pressure", "10"), (4, "ID", "55C"), (3, "Resistance", "20"), (2, "Pressure", "5"), (1, "ID", "75B"))
val inputDf = spark.createDataFrame(d1).toDF("time", "name", "value")
inputDf.createOrReplaceTempView("t")
val sql = """
select time, name, value, MAX(time_tmp) OVER (ORDER BY time ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) as ID
from (
  select *, CASE name WHEN "ID" THEN time ELSE -1 END as time_tmp
  from t
) as t2
"""
val outputDf = spark.sql(sql)

inputDf.show()
outputDf.show()

字符串
输出量：

+----+----------+-----+
|time|      name|value|
+----+----------+-----+
|   5|  Pressure|   10|
|   4|        ID|  55C|
|   3|Resistance|   20|
|   2|  Pressure|    5|
|   1|        ID|  75B|
+----+----------+-----+

+----+----------+-----+---+
|time|      name|value| ID|
+----+----------+-----+---+
|   1|        ID|  75B|  1|
|   2|  Pressure|    5|  1|
|   3|Resistance|   20|  1|
|   4|        ID|  55C|  4|
|   5|  Pressure|   10|  4|
+----+----------+-----+---+

型