title: How to detect when a pattern changes in a pyspark dataframe column
body: <p>I have a dataframe like below:</p>
  
  <pre><code>+-------------------+--------+-----------+
  |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         |
  +-------------------+--------+-----------+
  </code></pre>
  
  <p>I want the 'DateTime' value when values of result changes in each cycle. So basically 30 to 2A is a cycle for each UID. Now for some cases there <strong>can be data miss</strong>, in that case have to populate "datamiss",for example for a cycle(30-2A) if there is no record for '29' then in the below <strong>1st_chnage</strong> column should be "datamiss". For each unique result I have to take the last occurrence  EXCEPT the very first record for every cycle</p>
  
  <p>Based on this I want a output like this:</p>
  
  <pre><code>|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|
  </code></pre>
  
  <p>How can I do that in most efficient way considering I have to do this multiple times for each sensor id and the data set is having 1000k records.</p>
  
  <p>So far, I was able to do upto this but not able to get to the right point, unable to handle the dynamic ness when there is data miss</p>
  
  <pre><code>    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")))
  </code></pre>
  
  <p>Based on this I want a output like this:</p>
  
  <pre><code>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
  </code></pre>
  
  <p>How can I do that in most efficient way considering I have to do this multiple times for each sensor id and the data set is having 1000k records.</p>
  
  <p>So far, I was able to do upto this.</p>
  
  <pre><code>    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")))
  </code></pre>