Pyspark：按 ID 和最近日期向后连接 2 个数据帧答案

【问题标题】：Pyspark: Joining 2 dataframes by ID & Closest date backwardsPyspark：按 ID 和最近日期向后连接 2 个数据帧
【发布时间】：2020-08-08 02:31:16
【问题描述】：

我在 pyspark（和一般的 python）中执行两个数据帧的滚动连接时遇到了很多问题。我希望通过它们的 ID 和最接近的日期向后将两个 pyspark 数据帧连接在一起（意味着第二个数据帧中的日期不能大于第一个数据帧中的日期）

表_1：

表_2：

期望的结果：

本质上，我知道 SQL 查询可以完成我可以执行 spark.sql("query") 的技巧，所以其他任何事情。我已经尝试了一些在火花环境中不起作用的东西。谢谢！

【问题讨论】：

标签： python sql join pyspark

【解决方案1】：

这是我的试验。

首先，我确定符合您条件的Date_2。之后，再次加入第二个数据框，得到Value_2

from pyspark.sql.functions import monotonically_increasing_id, unix_timestamp, max

df3 = df1.withColumn('newId', monotonically_increasing_id()) \
  .join(df2, 'ID', 'left') \
  .where(unix_timestamp('Date', 'M/dd/yy') >= unix_timestamp('Date_2', 'M/dd/yy')) \
  .groupBy(*df1.columns, 'newId') \
  .agg(max('Date_2').alias('Date_2'))
df3.orderBy('newId').show(20, False)    

+---+-------+-----+-----+-------+
|ID |Date   |Value|newId|Date_2 |
+---+-------+-----+-----+-------+
|A1 |1/15/20|5    |0    |1/12/20|
|A2 |1/20/20|10   |1    |1/11/20|
|A3 |2/21/20|12   |2    |1/31/20|
|A1 |1/21/20|6    |3    |1/16/20|
+---+-------+-----+-----+-------+

df3.join(df2, ['ID', 'Date_2'], 'left') \
  .orderBy('newId') \
  .drop('Date_2', 'newId') \
  .show(20, False)

+---+-------+-----+-------+
|ID |Date   |Value|Value_2|
+---+-------+-----+-------+
|A1 |1/15/20|5    |5      |
|A2 |1/20/20|10   |12     |
|A3 |2/21/20|12   |14     |
|A1 |1/21/20|6    |3      |
+---+-------+-----+-------+

【讨论】：

【解决方案2】：

df1=spark.createDataFrame([('A1','1/15/2020',5),
                           ('A2','1/20/2020',10), 
                           ('A3','2/21/2020',12),
                           ('A1','1/21/2020',6)],
                           ['ID1','Date1','Value1'])

df2=spark.createDataFrame([('A1','1/10/2020',1),
                           ('A1','1/12/2020',5),
                           ('A1','1/16/2020',3),
                           ('A2','1/25/2020',20),
                           ('A2','1/1/2020',12),
                           ('A3','1/31/2020',14),
                           ('A3','1/30/2020',12)],['ID2','Date2','Value2'])

df2=df1.join(df2,df1.ID1==df2.ID2) \
    .withColumn("distance",datediff(to_date(df1.Date1,'MM/dd/yyyy'),\
     to_date(df2.Date2,'MM/dd/yyyy'))).filter("distance>0")

df2.groupBy(df2.ID1,df2.Date1,df2.Value1)\
   .agg(min(df2.distance).alias('distance')).join(df2, ['ID1','Date1','distance'])\
   .select(df2.ID1,df2.Date1,df2.Value1,df2.Value2).orderBy('ID1','Date1').show()

【讨论】：

我喜欢这个主意，但你能解释一下你在代码中做了什么吗？

【解决方案3】：

我更愿意使用 Window 来解决这个问题。
您需要使用 id 和 date(>=) 连接两个数据集，然后您需要知道相差多少天，以便使用 dense_rank 过滤您需要的内容，以获得最接近的日期。

from pyspark.sql.functions import col, datediff, dense_rank
from pyspark.sql.window import Window
from datetime import date

df1 = (
  spark
  .createDataFrame(
    [
      ("A1",date(2020, 1, 15), 5),
      ("A2",date(2020, 1, 20), 10),
      ("A3",date(2020, 2, 21), 12),
      ("A1",date(2020, 1, 21), 6),
    ],
    ["id_1","date_1","value_1"]
  )
)

df2 = (
  spark
  .createDataFrame(
    [
      ("A1",date(2020, 1, 10), 1),
      ("A1",date(2020, 1, 12), 5),
      ("A1",date(2020, 1, 16), 3),
      ("A2",date(2020, 1, 25), 20),
      ("A2",date(2020, 1, 1), 12),
      ("A3",date(2020, 1, 31), 14),
      ("A3",date(2020, 1, 30), 12)
    ],
    ["id_2","date_2","value_2"]
  )
)

winSpec = Window.partitionBy("value_1").orderBy("date_difference")

df3 = (
  df1
  .join(df2, [df1.id_1==df2.id_2,df1.date_1>=df2.date_2])
  .withColumn("date_difference", datediff("date_1","date_2"))
  .withColumn("dr", dense_rank().over(winSpec))
  .where("dr=1")
  .select(
    col("id_1").alias("id"),
    col("date_1").alias("date"),
    col("value_1"),
    col("value_2")
  )
)

+---+----------+-------+-------+
|id |date      |value_1|value_2|
+---+----------+-------+-------+
|A1 |2020-01-21|6      |3      |
|A1 |2020-01-15|5      |5      |
|A2 |2020-01-20|10     |12     |
|A3 |2020-02-21|12     |14     |
+---+----------+-------+-------+

【讨论】：