Выделение всех строк между двумя конкретными строками в столбце DataFrame с помощью Scala Spark

Question

У меня есть DataFrame, и в нем есть столбец с именем Events.Итак, что я хочу сделать, это собрать все строки между значением события от строки «Начало события» до «Конец события».то есть все строки, когда «Начало события» пришло до того, как «Конец события» поступило в столбец.Это «начало события» - «конец события» будет выполняться несколько раз в столбце, и все остальные строки, которые поступают между «окончанием события» и «началом события», должны быть отброшены.

Используется .filter, но он дает только строки «Начало события» и «Конец события», но не строки между

val df1 = selectedDf.filter(($"Event" === "Event Start"))

val df2 = selectedDf.filter(($"Event" === "Event End"))

Данные в некоторой степени похожи на: enterописание изображения здесь

Answer 1

Надеюсь, что следующий пример справки:

val df = sc.parallelize(Seq(("MS 1","X"), ("MS 2", "Y"), ("MS 3", "X"), ("MS 4", "E"), ("MS 3", "E"))).toDF("ms", "event")

df.show

val fileter1DF = df.filter(col("event") === "X")

val filter2DF = df.filter(col("event") === "E")

val result = fileter1DF.as("x").join(filter2DF.as("e"), List("ms"), "left_outer").where(col("e.event").isNull).select(col("x.ms"))

result.show

Результаты:

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_131)
Type in expressions to have them evaluated.
Type :help for more information.

scala> val df = sc.parallelize(Seq(("MS 1","X"), ("MS 2", "Y"), ("MS 3", "X"), ("MS4 2", "E"), ("MS 3", "E"))).toDF("ms", "event")
df: org.apache.spark.sql.DataFrame = [ms: string, event: string]

scala> df.show
+-----+-----+
|   ms|event|
+-----+-----+
| MS 1|    X|
| MS 2|    Y|
| MS 3|    X|
| MS 4|    E|
| MS 3|    E|
+-----+-----+


scala> val fileter1DF = df.filter(col("event") === "X")
fileter1DF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [ms: string, event: string]

scala> val filter2DF = df.filter(col("event") === "E")
filter2DF: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [ms: string, event: string]

scala> val result = fileter1DF.as("x").join(filter2DF.as("e"), List("ms"), "left_outer").where(col("e.event").isNull).select(col("x.ms"))
result: org.apache.spark.sql.DataFrame = [ms: string]

scala> result.show
+----+
|  ms|
+----+
|MS 1|
+----+


scala>

Answer 2

попробуйте это:

val data = spark.read.csv("data/sort.csv").toDF("val")
data.show()

+---+
|val|
+---+
|AAA|
|ABC|
|ABD|
|ABF|
|DEF|
|DEZ|
|ZZZ|
+---+

data.filter('val >= "ABC" && 'val <= "DEF").show()
+---+
|val|
+---+
|ABC|
|ABD|
|ABF|
|DEF|
+---+