在具有特定状态(列值)的行上分布时间段

Question

>我有一个包含日志的 Pyspark 数据帧，每行对应于记录时系统的状态，以及一个组号。我想找出每个组处于不健康状态的时间段的长度。

例如，如果这是我的表：

TIMESTAMP | STATUS_CODE | GROUP_NUMBER
--------------------------------------
02:03:11  | healthy     | 000001
02:03:04  | healthy     | 000001
02:03:03  | unhealthy   | 000001
02:03:00  | unhealthy   | 000001
02:02:58  | healthy     | 000008
02:02:57  | healthy     | 000008
02:02:55  | unhealthy   | 000001
02:02:54  | healthy     | 000001
02:02:50  | healthy     | 000007
02:02:48  | healthy     | 000004

我想返回组 000001 的不正常时间段为 9 秒(从 02：02：55 到 02：03：04(。

其他群体也可能有不健康的时间段，我也想返回这些时间段。

由于具有相同状态的连续行的可能性，并且由于不同组的行穿插在一起，我正在努力寻找一种有效地执行此操作的方法。

我无法将 Pyspark 数据帧转换为 Pandas 数据帧，因为它太大了。

如何有效地确定这些时间段的长度？

非常感谢！

Answer 1

带有Spark-SQL解决方案的pyspark看起来像这样。

首先，我们创建示例数据集。除了数据集之外，我们还生成row_number字段分区，并按时间戳排序。然后我们将生成的数据帧注册为表，例如table1

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number
from pyspark.sql.functions import unix_timestamp
df = spark.createDataFrame([
('2017-01-01 02:03:11','healthy','000001'),
('2017-01-01 02:03:04','healthy','000001'),
('2017-01-01 02:03:03','unhealthy','000001'),
('2017-01-01 02:03:00','unhealthy','000001'),
('2017-01-01 02:02:58','healthy','000008'),
('2017-01-01 02:02:57','healthy','000008'),
('2017-01-01 02:02:55','unhealthy','000001'),
('2017-01-01 02:02:54','healthy','000001'),
('2017-01-01 02:02:50','healthy','000007'),
('2017-01-01 02:02:48','healthy','000004')
],['timestamp','state','group_id'])
df = df.withColumn('rownum', row_number().over(Window.partitionBy(df.group_id).orderBy(unix_timestamp(df.timestamp))))
df.registerTempTable("table1")

将数据帧注册为表 ( table1 ( 后。 所需的数据可以使用Spark-SQL计算如下

>>> spark.sql("""
... SELECT t1.group_id,sum((t2.timestamp_value - t1.timestamp_value)) as duration
... FROM
... (SELECT unix_timestamp(timestamp) as timestamp_value,group_id,rownum FROM table1 WHERE state = 'unhealthy') t1
... LEFT JOIN
... (SELECT unix_timestamp(timestamp) as timestamp_value,group_id,rownum FROM table1) t2
... ON t1.group_id = t2.group_id
... AND t1.rownum = t2.rownum - 1
... group by t1.group_id
... """).show()
+--------+--------+
|group_id|duration|
+--------+--------+
|  000001|       9|
+--------+--------+

示例日期集仅包含group_id 00001的不正常数据。 但此解决方案适用于状态不正常group_ids的其他情况。

Answer 2

一种简单的方法(可能不是最佳的(是：

1. 映射到以 GROUP_NUMBER 为 Key K 的[K,V]
2. 使用 repartitionAndSortWithinPartitions ，因此您将在同一分区中拥有每个组的所有数据，并按TIMESTAMP排序。详细解释它是如何工作的在这个答案中： Pyspark：使用repartitionAndSortWithinPartitions与多个排序Critiria。
3. 最后，使用 mapPartitions 对单个分区中的排序数据进行迭代，以便您可以轻松找到所需的答案。(mapPartitions解释：pyspark mapPartitions函数如何工作？