使用时间间隔透视表答案

【问题标题】：Pivoting the table with Time intervals使用时间间隔透视表
【发布时间】：2022-11-08 00:01:46
【问题描述】：

所以我有一个这样的pySpark DataFrame：

+------+--------------+---------------+----------+----------+
|A_ID  |B_ID          |C_ID           | BEGIN    |   END    |
+------+--------------+---------------+----------+----------+
| 55264|     12       |         4     |2001-01-01|2012-08-05|
| 54897|     12       |         4     |2001-11-01|2012-11-30|
|  8222|     12       |         5     |2001-08-01|2012-12-31|
| 25001|     12       |         4     |2001-10-01|2015-08-30|
| 40001|     12       |         5     |2001-11-01|2012-12-31|
|  8217|     12       |         5     |2001-05-01|2020-12-31|
| 40002|     12       |         5     |2001-05-01|2009-05-01|
| 25002|     12       |         4     |2001-10-01|2020-05-30|
|  8073|     13       |         3     |2002-05-05|2003-05-04|
...

我想用 Multiindexing 旋转 DataFrame。我想要两个垂直 ID - A_ID 和 B_ID。在水平线上应该是周，从最早的开始日期开始。值将基于 C_ID，其中 0 表示在特定周内没有任何 C_ID，1 表示 C_ID 在本周有一些值，2 表示 A_ID/B_ID 行在特定周内有多个唯一 C_ID。如果有一些关于 C_ID 过去所有日子的信息，那就太好了。

最后它可能看起来像这样：

+-----+-----+-----+-----+-----+
|    Weeks  | w1  | w2  | w3  | ....
+-----+-----+
|B_ID | A_ID|
+-----+-----+------------------
| 12  |55264|  0  |  1  |  1  |
|     |82226|  2  |  1  |  0  |
|     |80732|
|     |55264|
|     |40001|
|     |54897|       etc...
| 13  |80732|
|     |32444|
...

我怎样才能做到？

【问题讨论】：

您如何创建周 - w1，w2，w3，...？您使用的是哪个日期列？
我想过从begincolumn 开始，到end 结束。
那么第一张唱片的星期几？
特定 ID 组合的情况，如果在可能的第一周有 C_ID，则设置为 1。假设B_ID = 1、A_ID = 1、C_ID = 4，日期为1.1.2011 - 14.1.2011，然后在数据透视表中：B_ID = 1、A_ID = 1、w1 = 1、@ 987654332@ = 1, w3 = 0. 等等其他情况。

标签： dataframe pyspark time-series pivot pivot-table

【解决方案1】：

复制数据

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

cols = ["A_ID", "B_ID", "C_ID", "BEGIN", "END"]

data = [(55264, 12, 4, "2001-01-01", "2012-08-05"),
    (54897, 12, 4, "2001-11-01", "2012-11-30"),
    (8222,  12, 5, "2001-08-01", "2012-12-31"),
    (40001, 12, 5, "2001-11-01", "2012-12-31"),
    (8217,  12, 5, "2001-05-01", "2020-12-31"),
    (40002, 12, 5, "2001-05-01", "2009-05-01"),
    (25002, 12, 4, "2001-10-01", "2020-05-30"),
    (8073,  13, 3, "2002-05-05", "2003-05-04")]

df_data = (spark.createDataFrame(data, schema=cols)
            .select("A_ID", "B_ID", "C_ID", 
                     F.col("BEGIN").cast("date"),
                     F.col("END").cast("date")))

计算和旋转：

1）如果您希望列中的所有周都从第一个开始，请创建 df_weeks 并稍后在数据透视之前加入结果数据框。如果您想要编号的列，请使用 week_no 到 groupby 并在下面的 2) 中进行透视：

w = Window.orderBy("week")
df_weeks = (df_data
             .agg(F.min(F.date_trunc("week", "BEGIN")).cast("date").alias("start"), 
                  F.max(F.date_trunc("week", "END")).cast("date").alias("end"))
        .withColumn("week", F.explode(F.expr('sequence(start, end, interval 1 week)')))
        .withColumn("week_no", F.row_number().over(w))
       ).select("week", "week_no")

2）为BEGIN和END之间的每个日期创建行，用date_trunc截断以获取每个日期的周开始，按周分组，A_ID，B_ID并计算不同的C_ID值，加入df_dates，然后旋转并填充空值0：

df = (df_data
       .withColumn("dates", F.explode(F.expr('sequence(BEGIN, END, interval 1 week)')))
       .withColumn("week", F.date_trunc("week", "dates").cast("date"))
       .join(df_weeks, "week", "right")
        .groupBy("week", "A_ID", "B_ID")
        .agg(F.countDistinct("C_ID").alias("count"))
        .groupBy("A_ID", "B_ID")
        .pivot("week")
        .agg(F.first("count"))
        .fillna(0))

【讨论】：