在pyspark数据帧中对连续行进行分组

以下是一种方法:

将行集合为一组,其中一组是一组具有相同行的行 user_id 那是连续的( start_time 与上一个匹配 end_time )那么你可以用这个 group 做你的聚合。

实现这一点的一种方法是创建中间指示符列,告诉您用户是否已更改或时间不是连续的。然后对指标列执行累积和,以创建 组 .

例如:

import pyspark.sql.functions as f
from pyspark.sql import Window

w1 = Window.orderBy("start_time")
df = df.withColumn(
        "userChange",
        (f.col("user_id") != f.lag("user_id").over(w1)).cast("int")
    )\
    .withColumn(
        "timeChange",
        (f.col("start_time") != f.lag("end_time").over(w1)).cast("int")
    )\
    .fillna(
        0,
        subset=["userChange", "timeChange"]
    )\
    .withColumn(
        "indicator",
        (~((f.col("userChange") == 0) & (f.col("timeChange")==0))).cast("int")
    )\
    .withColumn(
        "group",
        f.sum(f.col("indicator")).over(w1.rangeBetween(Window.unboundedPreceding, 0))
    )
df.show()
#+-------+----------+--------+--------+----------+----------+---------+-----+
#|user_id|start_time|end_time|duration|userChange|timeChange|indicator|group|
#+-------+----------+--------+--------+----------+----------+---------+-----+
#|      1|  19:00:00|19:30:00|      30|         0|         0|        0|    0|
#|      1|  19:30:00|19:40:00|      10|         0|         0|        0|    0|
#|      1|  19:40:00|19:43:00|       3|         0|         0|        0|    0|
#|      2|  20:00:00|20:10:00|      10|         1|         1|        1|    1|
#|      1|  20:05:00|20:15:00|      10|         1|         1|        1|    2|
#|      1|  20:15:00|20:35:00|      20|         0|         0|        0|    2|
#+-------+----------+--------+--------+----------+----------+---------+-----+

现在我们有了 组 列,我们可以聚合如下以获得所需的结果:

df.groupBy("user_id", "group")\
    .agg(
        f.min("start_time").alias("start_time"),
        f.max("end_time").alias("end_time"),
        f.sum("duration").alias("duration")
    )\
    .drop("group")\
    .show()
#+-------+----------+--------+--------+
#|user_id|start_time|end_time|duration|
#+-------+----------+--------+--------+
#|      1|  19:00:00|19:43:00|      43|
#|      1|  20:05:00|20:35:00|      30|
#|      2|  20:00:00|20:10:00|      10|
#+-------+----------+--------+--------+

下面是从Pault的答案中得出的一个工作解决方案:

创建数据帧:

rdd = sc.parallelize([(1,"19:00:00", "19:30:00", 30), (1,"19:30:00", "19:40:00", 10),(1,"19:40:00", "19:43:00", 3), (2,"20:00:00", "20:10:00", 10), (1,"20:05:00", "20:15:00", 10),(1,"20:15:00", "20:35:00", 20)])

df = spark.createDataFrame(rdd, ["user_id", "start_time", "end_time", "duration"])

df.show()

+-------+----------+--------+--------+
|user_id|start_time|end_time|duration|
+-------+----------+--------+--------+
|      1|  19:00:00|19:30:00|      30|
|      1|  19:30:00|19:40:00|      10|
|      1|  19:40:00|19:43:00|       3|
|      1|  20:05:00|20:15:00|      10|
|      1|  20:15:00|20:35:00|      20|
+-------+----------+--------+--------+

创建一个指示列,该列指示时间发生更改的时间,并使用累积和为每个组提供唯一的ID:

import pyspark.sql.functions as f
from pyspark.sql import Window

w1 =  Window.partitionBy('user_id').orderBy('start_time')
df = df.withColumn(
        "indicator",
        (f.col("start_time") != f.lag("end_time").over(w1)).cast("int")
    )\
    .fillna(
        0,
        subset=[ "indicator"]
    )\
    .withColumn(
        "group",
        f.sum(f.col("indicator")).over(w1.rangeBetween(Window.unboundedPreceding, 0))
    )
df.show()

+-------+----------+--------+--------+---------+-----+
|user_id|start_time|end_time|duration|indicator|group|
+-------+----------+--------+--------+---------+-----+
|      1|  19:00:00|19:30:00|      30|        0|    0|
|      1|  19:30:00|19:40:00|      10|        0|    0|
|      1|  19:40:00|19:43:00|       3|        0|    0|
|      1|  20:05:00|20:15:00|      10|        1|    1|
|      1|  20:15:00|20:35:00|      20|        0|    1|
+-------+----------+--------+--------+---------+-----+

现在groupby on user id和group变量。

+-------+----------+--------+--------+
|user_id|start_time|end_time|duration|
+-------+----------+--------+--------+
|      1|  19:00:00|19:43:00|      43|
|      1|  20:05:00|20:35:00|      30|
+-------+----------+--------+--------+