обрезать фрейм данных в искре, используя последнее появление значения в столбце

Question

У меня есть датафрейм, где я хочу обрезать его по последнему появлению значения Good в столбце PDP.Это для рассмотрения строк 5 и ниже.Все, что находится выше строки 5, не имеет значения.

+------+----+
|custId| PDP|
|  1001| New|
|  1002|Good|
|  1003| New| 
|  1004| New|
|  1005|Good|
|  1006| New|
|  1007| New|
|  1008| New|
|  1009| New|
+------+----+

Что мне нужно, это этот фрейм данных.Так как последнее Good действие произошло в строке 5

+------+----+
|custId| PDP|
|  1001| New|
|  1002|Good|
|  1003| New| 
|  1004| New|
|  1005|Good|
+------+----+

Answer 1

Вы должны найти индекс последней строки с Good в столбце PDP, а затем отфильтровать только те строки, которые меньше этого индекса.

custId

Если ваш custIdстолбец содержит увеличивающиеся идентификаторы в отсортированном порядке, тогда вы можете сделать следующее

import org.apache.spark.sql.functions._
val maxIdToFilter = df.filter(lower(col("PDP")) === "good").select(max(col("custId").cast("long"))).first().getLong(0)
df.filter(col("custId") <= maxIdToFilter).show(false)

monotically_increasing_id

Если ваш custId не отсортирован и в порядке возрастания, то вы можете использовать следующую логику

import org.apache.spark.sql.functions._
val dfWithRow = df.withColumn("rowNo", monotonically_increasing_id())
val maxIdToFilter = dfWithRow.filter(lower(col("PDP")) === "good").select(max("rowNo")).first().getLong(0)
dfWithRow.filter(col("rowNo") <= maxIdToFilter).drop("rowNo").show(false)

Надеюсь, ответ полезен

Answer 2

Вы можете попробовать:

df
  .filter($"PDP" === "Good")             // Filter good
  .select(max("custId").alias("maxId"))  // Find max id
  .crossJoin(df)
  .where($"custId" <= $"maxId")          // Filter records with id <= lastGoodId
  .drop("maxId")                         // Remove obsolete column