Python / SQL - 根据现有填充记录的回滚/前滚循环填充记录

Question

我在两张表中有一些数据，一张是日期列表（带有其他字段），从 2014 年 1 月 1 日到昨天。另一张表包含 2020 年一年的数值数据（系数/指标数据）。

日期表上两个数据集之间的左连接导致所有日期都被带回，仅填充 2020 年的数据年份，其余为空。

我想要做的是用 2020 年的数据在 -364 天的地图上填充到 2014 年（和未来）的历史。

例如

#+----------+-----------+
#|date      |metric     |
#+----------+-----------+
#|03/02/2018|null       |
#|04/02/2018|null       |
#|05/02/2018|null       |
#|06/02/2018|null       |
#|07/02/2018|null       |
#|08/02/2018|null       |
#|09/02/2018|null       |
#|10/02/2018|null       |
#|....      |           |
#|02/02/2019|null       |
#|03/02/2019|null       |
#|04/02/2019|null       |
#|05/02/2019|null       |
#|06/02/2019|null       |
#|07/02/2019|null       |
#|08/02/2019|null       |
#|09/02/2019|null       |
#|...       |...        |
#|01/02/2020|0.071957531|
#|02/02/2020|0.086542975|
#|03/02/2020|0.023767137|
#|04/02/2020|0.109725808|
#|05/02/2020|0.005774458|
#|06/02/2020|0.056242301|
#|07/02/2020|0.086208715|
#|08/02/2020|0.010676928|   

这就是我想要达到的目标：

#+----------+-----------+
#|date      |metric     |
#+----------+-----------+
#|03/02/2018|0.071957531|
#|04/02/2018|0.086542975|
#|05/02/2018|0.023767137|
#|06/02/2018|0.109725808|
#|07/02/2018|0.005774458|
#|08/02/2018|0.056242301|
#|09/02/2018|0.086208715|
#|10/02/2018|0.010676928|
#|....      |           |
#|02/02/2019|0.071957531|
#|03/02/2019|0.086542975|
#|04/02/2019|0.023767137|
#|05/02/2019|0.109725808|
#|06/02/2019|0.005774458|
#|07/02/2019|0.056242301|
#|08/02/2019|0.086208715|
#|09/02/2019|0.010676928|
#|...       |...        |
#|01/02/2020|0.071957531|
#|02/02/2020|0.086542975|
#|03/02/2020|0.023767137|
#|04/02/2020|0.109725808|
#|05/02/2020|0.005774458|
#|06/02/2020|0.056242301|
#|07/02/2020|0.086208715|
#|08/02/2020|0.010676928|   

值得注意的是，我最终可能不得不回到 2014 年以上，因此对人口的任何活力都会有所帮助！

我在 databricks 中执行此操作，因此我可以使用各种语言，但希望专注于 Python/Pyspark/SQL 解决方案。

任何帮助将不胜感激。

谢谢。 计算机断层扫描

Answer 1

首先创建新列month和year：

df_with_month = df.withColumn("month", f.month(f.to_timestamp("date", "dd/MM/yyyy")))
                  .withColumn("year", f.month(f.to_timestamp("date", "dd/MM/yyyy")))

import pyspark.sql.functions as f

使用 2020 年的数据创建一个新的 DataFrame：

df_2020 = df_with_month.filter(col("year") == 2020)
                       .withColumnRenamed("metric", "new_metric")

加入当月结果：

df_with_metrics = df_with_month.join(df_2020, df_with_month.month == df_2020.month, "left")
                               .drop("metric")
                               .withColumnRenamed("new_metric", "metric")

Answer 2

首先你可以添加时间戳列：

df = df.select(F.to_timestamp("date", "dd/MM/yyyy").alias('ts'), '*')

然后你可以在相同的月份和日期加入：

cond = [F.dayofmonth(F.col('left.ts')) == F.dayofmonth(F.col('right.ts')),
        F.month(F.col('left.ts')) == F.month(F.col('right.ts'))]

df.select('ts', 'date').alias('left').\
join(df.filter(F.year('ts')==2020).select('ts', 'metric').alias('right'), cond)\
.orderBy(F.col('left.ts')).drop('ts').show()

Answer 3

您可以使用日期差为 364 天的倍数的条件进行自联接：

import pyspark.sql.functions as F

df2 = df.join(
    df.toDF('date2', 'metric2'),
    F.expr("""
        datediff(to_date(date, 'dd/MM/yyyy'), to_date(date2, 'dd/MM/yyyy')) % 364 = 0
        and
        to_date(date, 'dd/MM/yyyy') <= to_date(date2, 'dd/MM/yyyy')
    """)
).select(
    'date',
    F.coalesce('metric', 'metric2').alias('metric')
).filter('metric is not null')

df2.show(999)
+----------+-----------+
|      date|     metric|
+----------+-----------+
|03/02/2018|0.071957531|
|04/02/2018|0.086542975|
|05/02/2018|0.023767137|
|06/02/2018|0.109725808|
|07/02/2018|0.005774458|
|08/02/2018|0.056242301|
|09/02/2018|0.086208715|
|10/02/2018|0.010676928|
|02/02/2019|0.071957531|
|03/02/2019|0.086542975|
|04/02/2019|0.023767137|
|05/02/2019|0.109725808|
|06/02/2019|0.005774458|
|07/02/2019|0.056242301|
|08/02/2019|0.086208715|
|09/02/2019|0.010676928|
|01/02/2020|0.071957531|
|02/02/2020|0.086542975|
|03/02/2020|0.023767137|
|04/02/2020|0.109725808|
|05/02/2020|0.005774458|
|06/02/2020|0.056242301|
|07/02/2020|0.086208715|
|08/02/2020|0.010676928|
+----------+-----------+