如何检测 pyspark 数据框列中的模式何时发生变化

How to detect when a pattern changes in a pyspark dataframe column

我有如下数据框:

+-------------------+--------+-----------+
|DateTime           |UID.    |result     |
+-------------------+--------+-----------+
|2020-02-29 11:42:34|0000111D|30         |
|2020-02-30 11:47:34|0000111D|30         |
|2020-02-30 11:48:34|0000111D|30         |
|2020-02-30 11:49:34|0000111D|30         |
|2020-02-30 11:50:34|0000111D|30         |
|2020-02-25 11:50:34|0000111D|29         |
|2020-02-25 11:50:35|0000111D|29         |
|2020-02-26 11:52:35|0000111D|29         |
|2020-02-27 11:52:35|0000111D|29         |
|2020-02-28 11:52:35|0000111D|29         |
|2020-03-01 11:52:35|0000111D|28         |
|2020-03-02 11:12:35|0000111D|28         |
|2020-03-02 11:52:35|0000111D|28         |
|2020-03-03 12:32:35|0000111D|28         |
|2020-03-04 12:02:35|0000111D|28         |
|2020-03-05 11:12:45|0000111D|28         |
|2020-03-06 11:02:45|0000111D|27         |
|2020-03-07 10:32:45|0000111D|27         |
|2020-03-08 11:52:45|0000111D|27         |
|2020-03-09 11:12:45|0000111D|27         |
|2020-03-10 11:12:45|0000111D|27         |
|2020-03-11 11:48:45|0000111D|27         |
|2020-03-12 11:02:45|0000111D|27         |
|2020-03-13 11:28:45|0000111D|26         |
|2020-03-14 11:12:45|0000111D|26         |
|2020-03-15 11:12:45|0000111D|26         |
|2020-03-16 11:28:45|0000111D|26         |
|2020-03-17 11:42:45|0000111D|26         |
|2020-03-18 11:32:45|0000111D|26         |
|2020-03-19 11:28:45|0000111D|26         |
|2020-03-27 11:28:45|0000111D|2A         |
|2020-04-20 11:12:45|0000111D|2A         |
|2020-04-27 11:15:45|0000111D|2A         |
|2020-04-28 12:17:45|0000111D|2A         |
|2020-04-29 12:17:45|0000111D|30         |
|2020-04-30 12:18:45|0000111D|30         |
|2020-04-25 12:19:45|0000111D|30         |
|2020-04-26 12:20:45|0000111D|29         |
|2020-04-27 12:27:45|0000111D|29         |
|2020-04-28 12:28:45|0000111D|29         |
|2020-04-29 12:29:45|0000111D|28         |
|2020-05-01 12:26:45|0000111D|28         |
|2020-05-02 12:26:45|0000111D|27         |
|2020-05-03 12:26:45|0000111D|27         |
|2020-05-03 12:27:45|0000111D|26         |
|2020-05-05 12:29:45|0000111D|26         |
|2020-05-07 12:30:45|0000111D|2A         |
|2020-05-08 12:33:45|0000111D|2A         |
|2020-05-09 12:26:45|0000111D|2A         |
|2020-05-12 12:26:45|0000111D|30         |
|2020-05-14 11:52:35|0000111D|29         |
|2020-05-16 11:52:35|0000111D|28         |
|2020-05-18 11:52:35|0000111D|27         |
|2020-05-20 11:52:35|0000111D|26         |
|2020-05-27 11:52:35|0000111D|2A         |
+-------------------+--------+-----------+

我想要 'DateTime' 每个循环中结果值变化时的值。所以基本上30到2A是每个UID一个周期。现在在某些情况下 可能会丢失数据 ,在这种情况下必须填充 "datamiss",例如,如果没有 '29 的记录,则循环 (30-2A) ' 那么下面的1st_chnage栏应该是"datamiss"。对于每个唯一的结果,除了每个周期的第一个记录之外,我必须取最后一次出现的结果

基于此我想要这样的输出:

|UID     |        start_point|         1st_change|         2nd_change|         3rd_change|         4th_change|         5th_change|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+
|0000111D|2020-02-29 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:28:45|2020-04-28 12:17:45|
|0000111D|2020-04-29 12:17:45|2020-04-28 12:28: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|
|0000111D|2020-05-12 12:26:45|2020-05-14 11:52:35|2020-05-16 11:52:35|2020-05-18 11:52:35|2020-05-20 11:52:35|2020-05-27 11:52:35|

考虑到我必须为每个传感器 ID 多次执行此操作并且数据集有 1000k 条记录,我怎样才能以最有效的方式执行此操作。

到目前为止,我能够做到这一点,但无法到达正确的位置,无法处理数据缺失时的动态性

    w = Window.orderBy("DateTime")
    df_temp1=df.withColumn("rn",row_number().over(w)).\
    withColumn("lead",lead(col("result"),1).over(w)).\
    withColumn("lag",lag(col("result"),1).over(w)).withColumn("mismatch_bool",when((col('lead') != col('lag')),lit("true")).otherwise(lit("False")))

基于此我想要这样的输出:

sensorid  start_point         1st_change          2nd_change           3rd chnage          4th_change           5th chnage
0000126D  2020-02-23 11:42:34 2020-02-24 11:49:34 2020-02-25 11:52:34  2020-02-26 11:34:35 2020-02-28 11:43:35  null
0000126D  2020-03-01 11:23:35 2020-03-04 11:31:35 2020-03-06 11:17:35  2020-03-08 09:34:09 2020-03-10 11:34:09  2020-03-08 07:34:09

考虑到我必须为每个传感器 ID 多次执行此操作并且数据集有 1000k 条记录,我怎样才能以最有效的方式执行此操作。

到目前为止,我能做到这一点。

    w = Window.orderBy("DateTime")
    df_temp1=df_records_indiv_sensor.withColumn("rn",row_number().over(w)).\
    withColumn("lead",lead(col("result"),1).over(w)).\
    withColumn("lag",lag(col("result"),1).over(w)).withColumn("mismatch_bool",when((col('lead') != col('lag')),lit("true")).otherwise(lit("False")))

Spark2.4 only.

不确定这是否是您想要的,但我还是写了它,所以认为 id post it。这里有两个真正的挑战。 First 是在 30-2A 的数据中创建分区,并能够在这些分区中找到所需的更改。 Second,就是处理缺失行,只发送到缺失行的区间。(使用sequence[解决=39=] 等)。

这整个代码可能不是您想要的(我可能有点忘乎所以),但您可以 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 逻辑而不是序列处理数据丢失。

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|
+--------+-------------------+-------------------+-------------------+-------------------+-------------------+-------------------+