Grab last different data on Spark Dataframe?

Question

I have this data on Spark Dataframe

+------+-------+-----+------------+----------+---------+
|sernum|product|state|testDateTime|testResult|      msg|
+------+-------+-----+------------+----------+---------+
|     8|    PA1|  1.0|        1.18|      pass|testlog18|
|     7|    PA1|  1.0|        1.17|      fail|testlog17|
|     6|    PA1|  1.0|        1.16|      pass|testlog16|
|     5|    PA1|  1.0|        1.15|      fail|testlog15|
|     4|    PA1|  2.0|        1.14|      fail|testlog14|
|     3|    PA1|  1.0|        1.13|      pass|testlog13|
|     2|    PA1|  2.0|        1.12|      pass|testlog12|
|     1|    PA1|  1.0|        1.11|      fail|testlog11|
+------+-------+-----+------------+----------+---------+

What I care about is the testResult == "fail", and the hard part is that I need the to get the last "pass" message as an extra column GROUP BY product+state:

+------+-------+-----+------------+----------+---------+---------+
|sernum|product|state|testDateTime|testResult|      msg|  passMsg|
+------+-------+-----+------------+----------+---------+---------+
|     7|    PA1|  1.0|        1.17|      fail|testlog17|testlog16|
|     5|    PA1|  1.0|        1.15|      fail|testlog15|testlog13|
|     4|    PA1|  2.0|        1.14|      fail|testlog14|testlog12|
|     1|    PA1|  1.0|        1.11|      fail|testlog11|     null|
+------+-------+-----+------------+----------+---------+---------+

How can I achieve this using DataFrame or SQL?

Answer 1

The trick is to define groups where each group starts with a passed test. Then, use again window-functions with group as an additional partition-column:

val df = Seq(
  (8, "PA1", 1.0, 1.18, "pass", "testlog18"),
  (7, "PA1", 1.0, 1.17, "fail", "testlog17"),
  (6, "PA1", 1.0, 1.16, "pass", "testlog16"),
  (5, "PA1", 1.0, 1.15, "fail", "testlog15"),
  (4, "PA1", 2.0, 1.14, "fail", "testlog14"),
  (3, "PA1", 1.0, 1.13, "pass", "testlog13"),
  (2, "PA1", 2.0, 1.12, "pass", "testlog12"),
  (1, "PA1", 1.0, 1.11, "fail", "testlog11")
).toDF("sernum", "product", "state", "testDateTime", "testResult", "msg")


df
  .withColumn("group", sum(when($"testResult" === "pass", 1)).over(Window.partitionBy($"product", $"state").orderBy($"testDateTime")))
  .withColumn("passMsg", when($"group".isNotNull,first($"msg").over(Window.partitionBy($"product", $"state", $"group").orderBy($"testDateTime"))))
  .drop($"group")
  .where($"testResult"==="fail")
  .orderBy($"product", $"state", $"testDateTime")
  .show()

+------+-------+-----+------------+----------+---------+---------+
|sernum|product|state|testDateTime|testResult|      msg|  passMsg|
+------+-------+-----+------------+----------+---------+---------+
|     7|    PA1|  1.0|        1.17|      fail|testlog17|testlog16|
|     5|    PA1|  1.0|        1.15|      fail|testlog15|testlog13|
|     4|    PA1|  2.0|        1.14|      fail|testlog14|testlog12|
|     1|    PA1|  1.0|        1.11|      fail|testlog11|     null|
+------+-------+-----+------------+----------+---------+---------+