Определить, когда значение определенного столбца изменяется от значения 1 до значения 2 в pyspark

Question

У меня есть датафрейм, как показано ниже:

|DateTime           |uid      |fid.    |code       |DataLen|result     |

|2020-02-23 11:42:34|38       |0000126D|35         |02     |24         |
|2020-02-24 11:47:34|38       |0000126D|35         |02     |24         |
|2020-02-24 11:48:34|38       |0000126D|35         |02     |23         |
|2020-02-24 11:49:34|38       |0000126D|35         |02     |23         |
|2020-02-24 11:50:34|38       |0000126D|35         |02     |22         |
|2020-02-25 11:52:34|38       |0000126D|35         |02     |22         |
|2020-02-25 11:12:35|38       |0000126D|35         |02     |21         |
|2020-02-26 11:34:35|38       |0000126D|35         |02     |21         |
|2020-02-27 11:12:35|38       |0000126D|35         |02     |2A         |
|2020-02-28 11:43:35|38       |0000126D|35         |02     |2A         |
|2020-03-01 11:23:35|38       |0000126D|35         |02     |24         |
|2020-03-02 11:10:35|38       |0000126D|35         |02     |23         |
|2020-03-03 11:07:35|38       |0000126D|35         |02     |22         |
|2020-03-04 11:31:35|38       |0000126D|35         |02     |21         |
|2020-03-05 11:07:35|38       |0000126D|35         |02     |2A         |
|2020-03-06 11:17:35|38       |0000126D|35         |02     |2A         |
|2020-03-07 11:15:47|38       |0000126D|35         |02     |24         |
|2020-03-08 11:34:09|38       |0000126D|35         |02     |24         |

Вывод, который мне нужен:

|DateTime           |uid      |fid.    |code       |DataLen|result     |Bool
|2020-02-23 11:42:34|38       |0000126D|35         |02     |24         |T0
|2020-02-24 11:47:34|38       |0000126D|35         |02     |24         |F
|2020-02-24 11:48:34|38       |0000126D|35         |02     |23         |F
|2020-02-24 11:49:34|38       |0000126D|35         |02     |23         |F
|2020-02-24 11:50:34|38       |0000126D|35         |02     |22         |F
|2020-02-25 11:52:34|38       |0000126D|35         |02     |22         |F
|2020-02-25 11:12:35|38       |0000126D|35         |02     |21         |F
|2020-02-26 11:34:35|38       |0000126D|35         |02     |21         |F
|2020-02-27 11:12:35|38       |0000126D|35         |02     |2A         |F
|2020-02-28 11:43:35|38       |0000126D|35         |02     |2A         |T1
|2020-03-01 11:23:35|38       |0000126D|35         |02     |24         |T0
|2020-03-02 11:10:35|38       |0000126D|35         |02     |23         |F
|2020-03-03 11:07:35|38       |0000126D|35         |02     |22         |F
|2020-03-04 11:31:35|38       |0000126D|35         |02     |21         |F
|2020-03-05 11:07:35|38       |0000126D|35         |02     |2A         |F
|2020-03-06 11:17:35|38       |0000126D|35         |02     |2A         |T1
|2020-03-07 11:15:47|38       |0000126D|35         |02     |24         |T0
|2020-03-08 11:34:09|38       |0000126D|35         |02     |24         |F

Я хочу получить значение DateTime, когда значения результата изменяются с 2A до 24. Таким образом, в основном от 24 до 2А - это цикл для каждого датчика. Я хочу получить значение datetime для первой записи «24» (T0) и когда значение меняется, как указано выше.

Как я могу сделать это наиболее эффективным способом, учитывая, что мне нужно сделать это несколько раз для каждого датчика идентификатор и набор данных имеет 1000 тыс. записей

Answer 1

Используйте .window lag,lead,row_number функций для этого случая.

Example:

from pyspark.sql.functions import *
from pyspark.sql import Window
w = Window.orderBy("datetime")

#if rn 1 and result=24 then add T0, if result 2A next value =24 then T1 else F
#if lag_bool T1 and next value is F then T0

df.withColumn("rn",row_number().over(w)).\
withColumn("lead",lead(col("result"),1).over(w)).\
withColumn("bool_tmp",when((col("result") == "24") &(col("rn") == 1) , lit("T0")).when((col("lead") == "24") & (col("result") =="2A"), lit("T1")).otherwise(lit("F"))).\
withColumn("lag_bool",lag(col("bool_tmp"),1).over(w)).withColumn("bool",when((col("bool_tmp") == "F") & (col("lag_bool") =="T1"), lit("T0")).otherwise(col("bool_tmp"))).\
drop("rn","bool_tmp","lag_bool","lead").\
show()

#+-------------------+---+--------+----+-------+------+----+
#|           datetime|uid|sensorid|code|datalen|result|bool|
#+-------------------+---+--------+----+-------+------+----+
#|2020-02-23 11:42:34| 38|0000126D|  35|     02|    24|  T0|
#|2020-02-24 11:47:34| 38|0000126D|  35|     02|    24|   F|
#|2020-02-24 11:48:34| 38|0000126D|  35|     02|    23|   F|
#|2020-02-24 11:49:34| 38|0000126D|  35|     02|    23|   F|
#|2020-02-24 11:50:34| 38|0000126D|  35|     02|    22|   F|
#|2020-02-25 11:12:35| 38|0000126D|  35|     02|    21|   F|
#|2020-02-25 11:52:34| 38|0000126D|  35|     02|    22|   F|
#|2020-02-26 11:34:35| 38|0000126D|  35|     02|    21|   F|
#|2020-02-27 11:12:35| 38|0000126D|  35|     02|    2A|   F|
#|2020-02-28 11:43:35| 38|0000126D|  35|     02|    2A|  T1|
#|2020-03-01 11:23:35| 38|0000126D|  35|     02|    24|  T0|
#|2020-03-02 11:10:35| 38|0000126D|  35|     02|    23|   F|
#|2020-03-03 11:07:35| 38|0000126D|  35|     02|    22|   F|
#|2020-03-04 11:31:35| 38|0000126D|  35|     02|    21|   F|
#|2020-03-05 11:07:35| 38|0000126D|  35|     02|    2A|   F|
#|2020-03-06 11:17:35| 38|0000126D|  35|     02|    2A|  T1|
#|2020-03-07 11:15:47| 38|0000126D|  35|     02|    24|  T0|
#|2020-03-08 11:34:09| 38|0000126D|  35|     02|    24|   F|
#+-------------------+---+--------+----+-------+------+----+