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>There is no efficient method which can perform operation like this out-of-the-box. You could use window functions:</p>

<pre><code>from pyspark.sql.window import Window
from pyspark.sql.functions import col, lag

df = sc.parallelize([
    (1473678846, 2), (1473678852, 1),
    (1473679029, 3), (1473679039, 3),
    (1473679045, 2), (1473679055, 1)
]).toDF(["time", "door_status"])


w = Window().orderBy("time")
(df
    .withColumn("prev_status", lag("door_status", 1).over(w))
    .where(col("door_status") != col("prev_status"))
    .groupBy("door_status", "prev_status")
    .count())
</code></pre>

<p>but this simply won't scale. You can try to <code>mapParitions</code>. Fist let's define a function we'll use to map partitions:</p>

<pre><code>from collections import Counter

def process_partition(iter):
    """Given an iterator of (time, state)
    return the first state, the last state and 
    a counter of state changes

    &gt;&gt;&gt; process_partition([])
    [(None, None, Counter())]
    &gt;&gt;&gt; process_partition(enumerate([1, 1, 1]))
    [(1, 1, Counter())]
    &gt;&gt;&gt; process_partition(enumerate([1, 2, 3]))
    [(1, 3, Counter({(1, 2): 1, (2, 3): 1}))]
    """

    first = None
    prev = None
    cnt  = Counter()

    for i, (_, x) in enumerate(iter):
        # Store the first object per partition
        if i == 0:
            first = x

        # If change of state update couter
        if prev is not None and prev != x:
            cnt[(prev, x)] += 1

        prev = x

    return [(first, prev, cnt)]
</code></pre>

<p>and a simple merger functions:</p>

<pre><code>def merge(x, y):
    """Given a pair of tuples:
    (first-state, last-state, counter_of changes)
    return a tuple of the same shape representing aggregated results

    &gt;&gt;&gt; merge((None, None, Counter()), (1, 1, Counter()))
    (None, 1, Counter())
    &gt;&gt;&gt; merge((1, 2, Counter([(1, 2)])), (2, 2, Counter()))
    (None, 2, Counter({(1, 2): 1}))
    &gt;&gt;&gt; merge((1, 2, Counter([(1, 2)])), (3, 2, Counter([(3, 2)])
    (None, 2, Counter({(1, 2): 1, (2, 3): 1, (3, 2): 1}))
    """
    (_, last_x, cnt_x), (first_y, last_y, cnt_y) = x, y

    # If previous partition wasn't empty update counter
    if last_x is not None and first_y is not None and last_x != first_y:
        cnt_y[(last_x, first_y)] += 1

    # Merge counters
    cnt_y.update(cnt_x)

    return (None, last_y, cnt_y)
</code></pre>

<p>With these two we can perform operation like this:</p>

<pre><code>partials = (df.rdd
    .mapPartitions(process_partition)
    .collect())

reduce(merge, [(None, None, Counter())] + partials)
</code></pre>

python apache-spark pyspark