我是 Apache Spark (Pyspark) 的新手，很高兴能得到一些帮助来解决这个问题。我目前正在使用 Pyspark 1.6(由于不支持 MQTT，不得不放弃 2.0)。
所以，我有一个包含以下信息的数据框，
+----------+-----------+
| time|door_status|
+----------+-----------+
|1473678846| 2|
|1473678852| 1|
|1473679029| 3|
|1473679039| 3|
|1473679045| 2|
|1473679055| 1|
这基本上就是门的时间和状态。我需要计算门打开和关闭的次数。所以我需要识别状态变化并为每个状态保留独立的计数器。
由于我是新手，所以我发现很难想象如何实现它。如果有人可以提出一个想法并指出我正确的方向，那就太好了。
提前致谢!
请您参考如下方法:
没有有效的方法可以立即执行这样的操作。您可以使用窗口函数:
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())
但这根本无法扩展。您可以尝试 mapParitions。首先让我们定义一个我们将用来映射分区的函数:
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
>>> process_partition([])
[(None, None, Counter())]
>>> process_partition(enumerate([1, 1, 1]))
[(1, 1, Counter())]
>>> 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)]
和一个简单的合并函数:
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
>>> merge((None, None, Counter()), (1, 1, Counter()))
(None, 1, Counter())
>>> merge((1, 2, Counter([(1, 2)])), (2, 2, Counter()))
(None, 2, Counter({(1, 2): 1}))
>>> 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)
有了这两个我们就可以执行这样的操作:
partials = (df.rdd
.mapPartitions(process_partition)
.collect())
reduce(merge, [(None, None, Counter())] + partials)
