使用Pyspark

Question

目前，我正在使用pyspark进行计算，并尝试在特定条件下从多个数据范围匹配数据。 我是Pyspark的新手，决定寻求帮助。 我的第一个数据框架包含有关贷款的一般信息：

ID     ContractDate MaturityDate Bank
ID1    2024-06-01   2024-06-18   A       
ID2    2024-06-05   2024-06-18   B       
ID3    2024-06-10   2024-06-17   C       
ID4    2024-06-15                D

ID5    2024-08-01   2024-08-22   A       
ID6    2024-08-08   2024-08-23   B            
ID7    2024-08-20                D

我的第二个数据框架包含有关付款方式的信息。

对于每笔贷款，我有一笔或多项付款。

ID_loan PaymentDate PaymentSum
ID1     2024-06-02  10
ID1     2024-06-08  40
ID1     2024-06-10  50
ID2     2024-06-06  30
ID2     2024-06-07  90
ID2     2024-06-08  20
ID3     2024-06-11  20
ID3     2024-06-12  30
ID3     2024-06-13  50

ID5     2024-08-10  15
ID5     2024-08-13  35
ID5     2024-08-15  30
ID6     2024-08-15  20
ID6     2024-08-16  20
ID6     2024-08-20  70

我的目标是将第一个数据框架添加到第一个数据框架“ Paymentsum”列，这将为每笔贷款返还付款金额，因为该付款是在银行发行的最接近的“合同日期”的最接近日期'd'

其他单词我必须得到下表：

ID     ContractDate MaturityDate Bank PaymentSum
ID1    2024-06-01   2024-06-18   A    50 
ID2    2024-06-05   2024-06-18   B    20   
ID3    2024-06-10   2024-06-17   C    50   
ID4    2024-06-15                D

ID5    2024-08-01   2024-08-22   A    30   
ID6    2024-08-08   2024-08-23   B    70       
ID7    2024-08-20                D

我确实知道加入这里还不够
任何帮助都将受到高度赞赏！

您需要在这种情况下使用窗口功能。

from pyspark.sql import SparkSession
from pyspark.sql import functions as func

df_1_ref = df_1.withColumn(
    "closest_contract_date_from_D", func.first(
        func.when(func.col("Bank")=="D", func.col("ContractDate")).otherwise(None), ignorenulls=True
    ).over(
        Window.orderBy(func.asc("ContractDate")).rowsBetween(Window.currentRow, Window.unboundedFollowing)
    )
).select(
    "ID", "closest_contract_date_from_D"
)

首先，您创建一个参考表以收集最接近银行发行的贷款的日期D。我在此处使用的方法是使用窗口函数与排名进行查找。对于每一行，它将仅从当前行搜索到未限制的行。

ContractDate

Answer 1

df_2 = df_2.join(
    df_1_ref, on=[func.col("ID")==func.col("ID_loan"), func.col("PaymentDate")<=func.col("closest_contract_date_from_D")], how="left"
).withColumn(
    "rank", func.rank().over(Window.partitionBy("ID_loan").orderBy(func.desc("PaymentDate")))
).filter(
    func.col("rank") == 1
).selectExpr(
    "ID_loan AS ID", "PaymentSum"
)

即可确保您只需要付款日期，而付款日期比每个记录中的参考较小。然后，您可以使用等级函数来查找最接近的功能。

<=然后您可以加入主要数据帧。