apachespark—基于下次Dataframe中出现特定值时更新行

jjjwad0x 于 2021-05-27 发布在 Spark

关注(0)|答案(1)|浏览(355)

如果我有这样的Dataframe

data = [(("ID1", "ENGAGEMENT", 2019-03-03)), (("ID1", "BABY SHOWER", 2019-04-13)), (("ID1", "WEDDING", 2019-07-10)), 
           (("ID1", "DIVORCE", 2019-09-26))]
    df = spark.createDataFrame(data, ["ID", "Event", "start_date"])
    df.show()

    +---+-----------+----------+
    | ID|      Event|start_date|
    +---+-----------+----------+
    |ID1| ENGAGEMENT|2019-03-03|
    |ID1|BABY SHOWER|2019-04-13|
    |ID1|    WEDDING|2019-07-10|
    |ID1|    DIVORCE|2019-09-26|
    +---+-----------+----------+

从这个Dataframe中，必须根据后续事件的开始日期推断事件的结束日期
例如：如果你有一个订婚，那么这将在婚礼结束时结束，所以你会把婚礼的开始日期作为订婚的结束日期。
所以上面的Dataframe应该得到这个输出。

+---+-----------+----------+----------+
| ID|      Event|start_date|  end_date|
+---+-----------+----------+----------+
|ID1| ENGAGEMENT|2019-03-03|2019-07-10|
|ID1|BABY SHOWER|2019-04-13|2019-04-13|
|ID1|    WEDDING|2019-07-10|2019-09-26|
|ID1|    DIVORCE|2019-09-26|      NULL|
+---+-----------+----------+----------+

我最初尝试在一个由id划分的窗口上使用lead函数来获取前面的行，但由于可能是20行之后，“婚礼”事件将是这样，它不起作用，而且是一种非常混乱的方式。

df = df.select("*", *([f.lead(f.col(c),default=None).over(Window.orderBy("ID")).alias("LEAD_"+c) 
                      for c in ["Event", "start_date"]]))

activity_dates = activity_dates.select("*", *([f.lead(f.col(c),default=None).over(Window.orderBy("ID")).alias("LEAD_"+c) 
                      for c in ["LEAD_Event", "LEAD_start_date"]]))

df = df.withColumn("end_date", f.when((col("Event") == "ENGAGEMENT") & (col("LEAD_Event") == "WEDDING"), col("LEAD_start_date"))
                                .when((col("Event") == "ENGAGEMENT") & (col("LEAD_LEAD_Event") == "WEDDING"), col("LEAD_LEAD_start_date"))

如何在不循环数据集的情况下实现这一点？

apache-spark pyspark apache-spark-sql pyspark-dataframes

来源：https://stackoverflow.com/questions/63855526/updating-rows-based-on-the-next-time-a-specific-value-occurs-in-a-dataframe-pysp

1条答案

按热度按时间

vjrehmav1#

这是我的尝试。

from pyspark.sql import Window
from pyspark.sql.functions import *

df.withColumn('end_date', expr('''
    case when Event = 'ENGAGEMENT'  then first(if(Event = 'WEDDING', start_date, null), True) over (Partition By ID)
         when Event = 'BABY SHOWER' then first(if(Event = 'BABY SHOWER', start_date, null), True) over (Partition By ID)
         when Event = 'WEDDING'     then first(if(Event = 'DIVORCE', start_date, null), True) over (Partition By ID)
    else null end
''')).show()

+---+-----------+----------+----------+
| ID|      Event|start_date|  end_date|
+---+-----------+----------+----------+
|ID1| ENGAGEMENT|2019-03-03|2019-07-10|
|ID1|BABY SHOWER|2019-04-13|2019-04-13|
|ID1|    WEDDING|2019-07-10|2019-09-26|
|ID1|    DIVORCE|2019-09-26|      null|
+---+-----------+----------+----------+

赞(0）回复(0）举报 2021-05-27

我来回答

apachespark—基于下次Dataframe中出现特定值时更新行

1条答案

相关问题

热门标签

最新问答