一种方法是创建一个"组"的中间 Dataframe ，告诉您每个组在哪一行开始和结束（类似于您已经做过的），然后使用大于/小于条件将其连接到原始表。
样本数据

scala> val input = Seq((1,"ST"),(2,"BPT"),(3,"SE"),(4,"ST"),(5,"BPT"),
                       (6,"N1"),(7,"SE"),(8,"ST"),(9,"PTD"),(10,"SE"))
                   .toDF("linenumber","segment")
scala> input.show(false)
+----------+-------+
|linenumber|segment|
+----------+-------+
|1         |ST     |
|2         |BPT    |
|3         |SE     |
|4         |ST     |
|5         |BPT    |
|6         |N1     |
|7         |SE     |
|8         |ST     |
|9         |PTD    |
|10        |SE     |
+----------+-------+

使用Window为组创建一个 Dataframe ，就像您的直觉告诉您的那样：

scala> val groups = input.where("segment='ST'")
                    .withColumn("endline",lead("linenumber",1) over Window.orderBy("linenumber"))
                    .withColumn("groupnumber",row_number() over Window.orderBy("linenumber"))
                    .withColumnRenamed("linenumber","startline")
                    .drop("segment")

scala> groups.show(false)
+---------+-----------+-------+
|startline|groupnumber|endline|
+---------+-----------+-------+
|1        |1          |4      |
|4        |2          |8      |
|8        |3          |null   |
+---------+-----------+-------+

将两者连接以获得结果

scala> input.join(groups,
                  input("linenumber") >= groups("startline") && 
                  (input("linenumber") < groups("endline") || groups("endline").isNull))
            .select("linenumber","segment","groupnumber") 
            .show(false)
+----------+-------+-----------+
|linenumber|segment|groupnumber|
+----------+-------+-----------+
|1         |ST     |1          |
|2         |BPT    |1          |
|3         |SE     |1          |
|4         |ST     |2          |
|5         |BPT    |2          |
|6         |N1     |2          |
|7         |SE     |2          |
|8         |ST     |3          |
|9         |PTD    |3          |
|10        |SE     |3          |
+----------+-------+-----------+

这样做的唯一问题是Window.orderBy()在未分区的 Dataframe 上，它会将所有数据收集到一个分区，因此可能是一个杀手。

如果您只想添加一个列，其编号在值“ST”出现在Segment列中时递增1，您可以在单独的 Dataframe 中过滤具有ST段的行，

var labelSTs = rawEDI.filter("segement == 'ST'");
// then group by ST and collect to list the linenumbers
var groupedDf = labelSTs.groupBy("Segment").agg(collect_list("Line_number").alias("Line_numbers"))
// now you need to flat back the data frame and log the line number index 
var flattedDf = groupedDf.select($"Segment", explode($"Line_numbers").as("Line_number"))   
// log the line_number index in your target column ST_Group
val withIndexDF = flattenedDF.withColumn("ST_Group", row_number().over(Window.partitionBy($"Segment").orderBy($"Line_number")))

结果是这样的：

+-------+----------+----------------+
|Segment|Line_number|ST_Group       |
+-------+----------+----------------+
|     ST|         1|               1|
|     ST|         4|               2|
|     ST|         8|               3|
+-------|----------|----------------|

然后将其与初始 Dataframe 中的其他段连接。

找到了一个更简单的方法，添加一列，当段列值为ST时，该列的值为1，否则该列的值为0。然后使用Window函数查找该新列的累积和。这将给您所需的结果。

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
val rawEDI=Seq((1,"ST"),(2,"BPT"),(3,"SE"),(4,"ST"),(5,"BPT"),(6,"N1"),(7,"SE"),(8,"ST"),(9,"PTD"),(10,"SE")).toDF("line_number","segment")

val newDf=rawEDI.withColumn("ST_Group", ($"segment" === "ST").cast("bigint"))

val windowSpec = Window.orderBy("line_number")
newDf.withColumn("ST_Group", sum("ST_Group").over(windowSpec))
.show

+-----------+-------+--------+
|line_number|segment|ST_Group|
+-----------+-------+--------+
|          1|     ST|       1|
|          2|    BPT|       1|
|          3|     SE|       1|
|          4|     ST|       2|
|          5|    BPT|       2|
|          6|     N1|       2|
|          7|     SE|       2|
|          8|     ST|       3|
|          9|    PTD|       3|
|         10|     SE|       3|
+-----------+-------+--------+