用以前的值填充日历表中的行

Question

我是 SQL 的新手，刚从 Python 和 R 转过来，并且将 Spark SQL 与 Databricks 结合使用。我正在尝试完成一个基本查询，希望得到指导，尤其是解释与我的问题相关的 SQL 基本概念的指导。

我有一个包含完整连续日期的日历表，以及一个包含date_added、user_id、sales 和price 列的数据表。数据表的日期不完整，因为并非每个用户在每个日期都处于活动状态。以下是每个表格的示例。

日历表

date
2020-01-01
2020-01-02
2020-01-03
2020-01-04
2020-01-05
2020-01-06

数据表

date_added     user_id    sales    price
2020-01-02     01         1        4.00
2020-01-05     01         3        4.00
2020-01-02     02         1        5.00
2020-01-03     02         1        5.00
2020-01-05     02         2        5.00
2020-01-03     03         2        1.00
2020-01-05     03         5        1.00

我正在寻找创建一个新表，其中为每个用户定义了特定范围内的每个日历表日期（活动日期），并且除销售列之外的所有列的空值是由该列中的以下值填充。大致如下：

date           user_id    sales    price         
2020-01-02     01         1        4.00
2020-01-03     01         null     4.00
2020-01-04     01         null     4.00
2020-01-05     01         3        4.00
2020-01-02     02         1        5.00
2020-01-03     02         1        5.00
2020-01-04     02         null     5.00
2020-01-05     02         2        5.00
2020-01-02     03         null     1.00
2020-01-03     03         2        1.00
2020-01-04     03         null     1.00
2020-01-05     03         5        1.00

感谢任何关于我如何进行此输出的指导。我尝试在日期上使用 LEFT JOIN，但没有成功。我知道 UNION 运算符用于将表连接在一起，但不知道如何在此处应用该方法。

Answer 1

您可以将用户与日历表交叉连接，然后与数据表左连接：

spark.sql("""
  SELECT  date, dates.user_id, sales, COALESCE(data.price, dates.price) AS price
  FROM    (
      SELECT  user_id, price, date
      FROM    (SELECT user_id, FIRST(price) as price FROM data_table GROUP BY user_id)
      CROSS JOIN calender_table
      WHERE   date >= (SELECT MIN(date_added) FROM data_table)
      AND     date <= (SELECT MAX(date_added) FROM data_table)
  )   dates
  LEFT JOIN data_table data
  ON      dates.user_id = data.user_id
  AND     dates.date = data.date_added
""").show()

输出：

+----------+-------+-----+-----+
|date      |user_id|sales|price|
+----------+-------+-----+-----+
|2020-01-02|01     |1    |4.0  |
|2020-01-03|01     |null |4.0  |
|2020-01-04|01     |null |4.0  |
|2020-01-05|01     |3    |4.0  |
|2020-01-02|02     |1    |5.0  |
|2020-01-03|02     |1    |5.0  |
|2020-01-04|02     |null |5.0  |
|2020-01-05|02     |2    |5.0  |
|2020-01-02|03     |null |1.0  |
|2020-01-03|03     |2    |1.0  |
|2020-01-04|03     |null |1.0  |
|2020-01-05|03     |5    |1.0  |
+----------+-------+-----+-----+

您还可以使用sequence 函数在不使用日历表的情况下生成日期。请参阅我的其他答案here。

Answer 2

让您的原始数据框为df1。然后您可以获取每个id 的min、max 日期并将其设为`df2'。

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

w = Window.partitionBy('user_id').orderBy(f.desc('date_added'))

df2 = df1.groupBy('user_id') \
  .agg(f.sequence(f.min('date_added'), f.max('date_added')).alias('date_added')) \
  .withColumn('date_added', f.explode('date_added'))

df2.join(df, ['user_id', 'date_added'], 'left') \
   .withColumn('price', f.first('price').over(w)) \
   .orderBy('user_id', 'date_added') \
   .show()

+-------+----------+-----+-----+
|user_id|date_added|sales|price|
+-------+----------+-----+-----+
|      1|2020-01-02|    1|  4.0|
|      1|2020-01-03| null|  4.0|
|      1|2020-01-04| null|  4.0|
|      1|2020-01-05|    3|  4.0|
|      2|2020-01-02|    1|  5.0|
|      2|2020-01-03|    1|  5.0|
|      2|2020-01-04| null|  5.0|
|      2|2020-01-05|    2|  5.0|
|      3|2020-01-03|    2|  1.0|
|      3|2020-01-04| null|  1.0|
|      3|2020-01-05|    5|  1.0|
+-------+----------+-----+-----+