Apache Spark Count State Change
<p>I am new to Apache Spark (Pyspark) and would be glad to get some help resolving this problem. I am currently using Pyspark 1.6 (Had to ditch 2.0 since there is not MQTT support).</p>

<p>So, I have a data frame that has the following information,</p>

<pre><code>+----------+-----------+
|      time|door_status|
+----------+-----------+
|1473678846|          2|
|1473678852|          1|
|1473679029|          3|
|1473679039|          3|
|1473679045|          2|
|1473679055|          1|
</code></pre>

<p>This is basically the time and status of the door. I need to calculate the number of times the door opened &amp; closed. So I need to identify the state change and keep independent counters for each status.</p>

<p>Since I am new to this I am finding it extremely hard to conceive any idea of how to implement this. If somebody can suggest an idea &amp; point me in the right direction it will be great. </p>

<p>Thanks in advance!!</p>

<p>You can try the following solution:</p>

<pre><code>import org.apache.spark.sql.expressions.Window

var data = Seq((1473678846, 2), (1473678852, 1), (1473679029, 3), (1473679039, 3), (1473679045, 2), (1473679055, 1),(1473779045, 1), (1474679055, 2), (1475679055, 1), (1476679055, 3)).toDF("time","door_status")

data.
  select(
    $"*",
    coalesce(lead($"door_status", 1).over(Window.orderBy($"time")), $"door_status").as("next_door_status")
  ).
  groupBy($"door_status").
  agg(
    sum(($"door_status" !== $"next_door_status").cast("int")).as("door_changes")
  ).
  show
</code></pre>

<p>It is in scala language. You have to craft the same in python.</p>

python apache-spark pyspark