Spark2.4 only.
Не уверен, что это то, что вы хотели бы, но я все равно написал это, так что я подумал опубликовать его . Здесь есть 2 реальных проблемы. First предназначено для получения создания разделов в данных, которые go из 30-2A, и возможности найти требуемые изменения в этих разделах. Second, для обработки отсутствующей строки, так что она отправляется только в интервал с отсутствующей строкой (решается с помощью sequence et c).
Весь этот код может быть не совсем тем, что вы хотите (возможно, я немного увлекся), но вы могли бы take parts of it and try them и , они могут помочь вам достичь нашей цели . .
В случае, если это именно то, что вы хотите, я объясню код более подробно. Но вы должны быть в состоянии выполнить большинство из них.
df.show()#your sample dataframe
+-------------------+--------+------+
| DateTime| UID|result|
+-------------------+--------+------+
|2020-02-23 11:42:34|0000111D| 30|
|2020-02-24 11:47:34|0000111D| 30|
|2020-02-24 11:48:34|0000111D| 29|
|2020-02-24 11:49:34|0000111D| 29|
|2020-02-24 11:50:34|0000111D| 28|
+-------------------+--------+------+
#only showing top 5 rows
from pyspark.sql import functions as F
from pyspark.sql.window import Window
w=Window().partitionBy("result").orderBy("DateTime")
w1=Window().partitionBy("UID").orderBy("DateTime")
w2=Window().partitionBy("UID","inc_sum").orderBy("DateTime")
w3=Window().partitionBy("UID","inc_sum")
w4=Window().partitionBy("DateTime","UID","inc_sum").orderBy("DateTime")
df.withColumn("cor",F.row_number().over(w))\
.withColumn("yo", F.when((F.col("cor")%2!=0) & (F.col("result")==30),F.lit(1)).otherwise(F.lit(0)))\
.withColumn("inc_sum", F.sum("yo").over(w1))\
.withColumn("cor", F.when((F.col("result")!=30) & (F.col("cor")%2==0), F.lit('change')).otherwise(F.lit('no')))\
.withColumn("row_num", F.row_number().over(w2))\
.withColumn("first", F.min("row_num").over(w3))\
.withColumn("max", F.max("row_num").over(w3)).drop("yo","row_num","first","max")\
.filter("row_num=first or row_num=max or cor='change'")\
.withColumn("all1", F.collect_list("result").over(w3))\
.withColumn("all", F.array(*[F.lit(x) for x in ['30','29','28','27','26','2A']]))\
.withColumn("except", F.array_except("all","all1")[0])\
.withColumn("result", F.when(F.col("except")+1==F.col("result"), F.expr("""sequence(int(except)+1,int(except),-1)"""))\
.otherwise(F.expr("""sequence(int(result),int(result),0)""")))\
.withColumn("result", F.when(F.col("result").isNull(), F.array(F.lit(2))).otherwise(F.col("result")))\
.select("DateTime","UID",F.explode("result").alias("result"),"inc_sum")\
.withColumn("rownum2", F.row_number().over(w4))\
.withColumn("DateTime", F.when((F.col("rownum2")>1), F.lit(0))\
.otherwise(F.col("DateTime"))).orderBy("DateTime")\
.groupBy("UID").pivot("result").agg((F.collect_list("DateTime")))\
.withColumn("zip", F.explode(F.arrays_zip(*['30','29','28','27','26','2'])))\
.select("UID", "zip.*")\
.select("UID", F.col("30").alias("start_point"),F.col("29").alias("1st_change"),F.col("28").alias("2nd_change")\
,F.col("27").alias("3rd_change"),F.col("26").alias("4th_change"),F.col("2").alias("5th_change"))\
.replace('0',"datamiss").show()
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
| UID| start_point| 1st_change| 2nd_change| 3rd_change| 4th_change| 5th_change|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
|0000111D|2020-02-23 11:42:34|2020-02-24 11:49:34|2020-02-25 11:52:34|2020-02-26 11:34:35| datamiss|2020-02-28 11:43:35|
|0000111D|2020-03-01 11:23:35|2020-03-04 11:31:35|2020-03-06 11:17:35|2020-03-08 11:34:09|2020-03-10 04:12:45|2020-03-12 07:34:09|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
UPDATED SOLUTION:
На основе новых предоставленных данных. Этот код был способен обрабатывать случаи, когда цикл запуска не всегда начинается с 24, и обрабатывал пропущенные данные, используя arrays_zip logi c вместо последовательности.
df.show()#new sample dataframe
+-------------------+---------+--------+-----------+-------+-----------+
| DateTime|Identity |UID Code |len |result|
+-------------------+---------+--------+-----------+-------+-----------+
|2020-02-25 11:50:34| 38|0000796D| 35| 2| 23|
|2020-02-25 11:50:35| 38|0000796D| 35| 2| 23|
|2020-02-26 11:52:35| 38|0000796D| 35| 2| 23|
|2020-02-27 11:52:35| 38|0000796D| 35| 2| 23|
|2020-02-28 11:52:35| 38|0000796D| 35| 2| 23|
+-------------------+---------+--------+-----------+-------+-----------+
#only showing top 5 rows
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.sql.functions import when
w=Window().partitionBy("UID").orderBy("DateTime")
w5=Window().partitionBy("UID","result","inc_sum").orderBy("DateTime")
w6=Window().partitionBy("UID","result","inc_sum")
w2=Window().partitionBy("UId","inc_sum").orderBy("DateTime")
w3=Window().partitionBy("UId","inc_sum")
w4=Window().partitionBy("DateTime","UId","inc_sum").orderBy("DateTime")
df.withColumn("lag", F.lag("result").over(w))\
.withColumn("lag", F.when(F.col("lag").isNull(),F.lit(-1)).otherwise(F.col("lag")))\
.withColumn("inc_sum", F.when((F.col("result")=='24')\
& (F.col("lag")!='24'),F.lit(1)).when((F.col("result")=='23')\
& (F.col("lag")!='24')&(F.col("lag")!='23'),F.lit(1)).otherwise(F.lit(0)))\
.withColumn("inc_sum", F.sum("inc_sum").over(w))\
.withColumn("row_num", F.row_number().over(w2))\
.withColumn("first", F.min("row_num").over(w3))\
.withColumn("max", F.max("row_num").over(w3))\
.withColumn("cor", F.row_number().over(w5))\
.withColumn("maxcor", F.max("cor").over(w6))\
.withColumn("maxcor", F.when((F.col("result")=='24') | (F.col("result")=='1F'), F.lit(None)).otherwise(F.col("maxcor"))).filter('row_num=first or row_num=max or cor=maxcor')\
.select("DateTime", "UID","result","inc_sum")\
.withColumn("result", F.when(F.col("result")=='1F', F.lit(19)).otherwise(F.col("result")))\
.withColumn("all1", F.collect_list("result").over(w3))\
.withColumn("all", F.array(*[F.lit(x) for x in ['24','23','22','21','20','19']]))\
.withColumn("except", F.when(F.size("all1")!=F.size("all"),F.array_except("all","all1")).otherwise(F.array(F.lit(None))))\
.withColumn("except2", F.flatten(F.array("all1","except")))\
.withColumn("except2", F.expr("""filter(except2,x-> x!='null')""")).drop("all1","all","except")\
.groupBy("UID","inc_sum").agg(F.collect_list("DateTime").alias("DateTime"),F.collect_list("result").alias("result")\
,F.first("except2").alias("except2"))\
.withColumn("zip", F.explode(F.arrays_zip("DateTime","result","except2")))\
.select("SensorId","zip.*","inc_sum")\
.withColumn("result", F.when(F.col("result").isNull(), F.col("except2")).otherwise(F.col("result")))\
.withColumn("DateTime", F.when(F.col("DateTime").isNull(), F.lit(0)).otherwise(F.col("DateTime")))\
.groupBy("UID").pivot("result").agg((F.collect_list("DateTime")))\
.withColumn("zipped", F.explode(F.arrays_zip(*['24','23','22','21','20','19'])))\
.select("UID", "zipped.*")\
.select("SensorId", F.col("24").alias("start_point"),F.col("23").alias("1st_change"),F.col("22").alias("2nd_change")\
,F.col("21").alias("3rd_change"),F.col("20").alias("4th_change"),F.col("19").alias("5th_change"))\
.replace('0',"datamiss").dropna()\
.show()
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
|UID. | start_point| 1st_change| 2nd_change| 3rd_change| 4th_change| 5th_change|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
|0000796D|2020-02-23 11:42:34|2020-02-28 11:52:35|2020-03-05 11:12:45|2020-03-12 11:02:45|2020-03-19 11:22:45|2020-04-22 12:17:45|
|0000796D|2020-05-12 12:26:45|2020-05-14 11:52:35|2020-05-16 11:52:35|2020-05-16 11:52:35|2020-05-20 11:52:35|2020-05-21 11:52:35|
|0000796D|2020-04-23 12:17:45|2020-04-28 12:22:45|2020-05-01 12:26:45|2020-05-03 12:26:45|2020-05-05 12:29:45|2020-05-09 12:26:45|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+