SQL / Pyspark - 根据动态时间戳和另一列添加新列答案

【问题标题】：SQL / Pyspark - Add new column based on a dynamic timestamp and another columnSQL / Pyspark - 根据动态时间戳和另一列添加新列
【发布时间】：2022-07-27 23:33:43
【问题描述】：

我有这些数据：

id, name, timestamp
1, David, 2022/01/01 10:00
2, David, 2022/01/01 10:30
3, Diego, 2022/01/01 10:59
4, David, 2022/01/01 10:59
5, David, 2022/01/01 11:01
6, Diego, 2022/01/01 12:00
7, David, 2022/01/01 12:00
8, David, 2022/01/01 12:05
9, Diego, 2022/01/01 12:30

基本上，大卫和迭戈正在玩游戏。他们不时在这些时间戳上按下按钮。

在他们第一次按下按钮后，游戏可以继续进行一小时。之后计数将重置，如果他们再次按下按钮，它将在他们再次开始播放时计数。

所以我想标记为0（开始），当他们在一个小时内第一次使用按钮时，如果他们在那个小时内，则标记为1（播放）。

所以在我的情况下，我会从结果中排除这个：

id, name, timestamp, status
1, David, 2022/01/01 10:00, 0  <--- David starts playing
2, David, 2022/01/01 10:30, 1  <--- David keeps playing the game that he started at the id 1
3, Diego, 2022/01/01 10:59, 0  <--- Diego starts playing
4, David, 2022/01/01 10:59, 1  <--- David keeps playing the game that he started at the id 1
5, David, 2022/01/01 11:01, 0  <--- David starts playing again
6, Diego, 2022/01/01 12:00, 0  <--- Diego starts playing again
7, David, 2022/01/01 12:00, 1  <--- David keeps playing the game that he started at the id 5
8, David, 2022/01/01 12:05, 0  <--- David start playing again
9, Diego, 2022/01/01 12:05, 1  <--- Diego keeps playing the game that he started at the id 6

我需要在 pyspark 中进行转换，以标记我认为的 start playing 和 keep playing。

如果你能帮我处理一个 SQL 查询，我最近可以把它改成 pyspark。

它不需要只在一个查询/步骤中完成。

希望你能帮助我。

【问题讨论】：

标签： sql pyspark

【解决方案1】：

这不是一个完整的解决方案，但有任何我尝试过的想法

from pyspark.sql.functions import explode
from datetime import datetime
from pyspark.sql.types import *
schema = StructType([StructField('id', StringType(), True),
                     StructField('name', StringType(), True),
                     StructField('timestamp', TimestampType(), True)])
df = spark.createDataFrame(
[
      ("1", "David", datetime.strptime("2022/01/01 10:00", '%Y/%m/%d %H:%M')),
      ("2", "David", datetime.strptime("2022/01/01 10:30",'%Y/%m/%d %H:%M')),
      ("3", "Diego", datetime.strptime("2022/01/01 10:59",'%Y/%m/%d %H:%M')),
      ("4", "David", datetime.strptime("2022/01/01 10:59", '%Y/%m/%d %H:%M')),
      ("5", "David", datetime.strptime("2022/01/01 11:01", '%Y/%m/%d %H:%M')),
      ("6", "Diego", datetime.strptime("2022/01/01 12:00", '%Y/%m/%d %H:%M')),
      ("7", "David", datetime.strptime("2022/01/01 12:00", '%Y/%m/%d %H:%M')),
      ("8", "David", datetime.strptime("2022/01/01 12:05", '%Y/%m/%d %H:%M')),
      ("9", "Diego", datetime.strptime("2022/01/01 12:30", '%Y/%m/%d %H:%M')),
],
schema=schema)
df.createOrReplaceTempView("people")
df3=spark.sql("select *,dense_rank()over(partition by hour(timestamp) order by name,timestamp )%2 as t4, case when dense_rank()over(partition by hour(timestamp) order by name,timestamp )%2>0 then dense_rank()over(partition by hour(timestamp) order by name,timestamp )%2-1 else  \
dense_rank()over(partition by hour(timestamp) order by name,timestamp )%2+1 end t3 from people order by timestamp,name")
df3.show()

【讨论】：