獲取最接近的匯率

我想根據第二個資料幀中的貨幣匯率和日期加入兩個資料幀。我已經嘗試過這里參考的方法，但是它datediff給出了日期的差異，所以它沒有給我正確的費率。

df1：

df2（貨幣表）：

輸出應如下所示：

唯一可行的方法是減去兩個日期“DATE”和“DATETra”之間的差異，并獲得最接近“DATETra”的日期。您能否提出另一種比減去字串更干凈的方法？

uj5u.com熱心網友回復：

您可以聚合較小的資料框 (df2)，以便將所有日期和費率收集到一個單元格中。然后，加入資料框，取出你需要的東西并進行除法。

輸入：

from pyspark.sql import functions as F
df1 = spark.createDataFrame(     
    [('AED', 'EUR', '2017-03-24', 2000),
     ('AED', 'EUR', '2017-03-27', 189),
     ('DZD', 'EUR', '2017-01-12', 130),
     ('EUR', 'EUR', '2020-01-01', 11)],
    ['from_curr', 'to_curr', 'Date', 'value_to_convert'])
df2 = spark.createDataFrame(
    [('AED', 'EUR', '2017-03-24', -5.123),
     ('AED', 'EUR', '2017-03-26', -9.5),
     ('DZD', 'EUR', '2017-01-01', -6.12)],
    ['transacti', 'local', 'DateTra', 'rate_exchange'])

獲取最接近日費率的腳本（可能來自未來）：

df2 = df2.groupBy('transacti', 'local').agg(
    F.collect_list(F.struct('DateTra', 'rate_exchange')).alias('_vals')
)
rate = F.array_sort(F.transform(
    '_vals',
    lambda x: F.struct(
        F.abs(F.datediff('Date', x.DateTra)).alias('diff'),
        -F.unix_timestamp(x.DateTra, 'yyyy-MM-dd').alias('DateTra'),
        F.abs(x.rate_exchange).alias('rate_exchange')
    )
))[0]['rate_exchange']
df = (df1
    .join(df2, (df1.from_curr == df2.transacti) & (df1.to_curr == df2.local), 'left')
    .select(
        df1['*'],
        F.coalesce(
            F.col('value_to_convert') / rate,
            F.when(df1.from_curr == df1.to_curr, df1.value_to_convert)
        ).alias('value_converted')
    )
)
df.show()
#  --------- ------- ---------- ---------------- ------------------ 
# |from_curr|to_curr|      Date|value_to_convert|   value_converted|
#  --------- ------- ---------- ---------------- ------------------ 
# |      AED|    EUR|2017-03-24|            2000| 390.3962521959789|
# |      AED|    EUR|2017-03-27|             189|19.894736842105264|
# |      EUR|    EUR|2020-01-01|              11|              11.0|
# |      DZD|    EUR|2017-01-12|             130|21.241830065359476|
#  --------- ------- ---------- ---------------- ------------------ 

獲取最新費率但不是來自未來的腳本：

df2 = df2.groupBy('transacti', 'local').agg(
    F.sort_array(F.collect_list(F.struct('DateTra', 'rate_exchange')), False).alias('_vals')
)

rate = F.abs(F.filter('_vals', lambda x: x.DateTra <= F.col('Date'))[0]['rate_exchange'])
df = (df1
    .join(df2, (df1.from_curr == df2.transacti) & (df1.to_curr == df2.local), 'left')
    .select(
        df1['*'],
        F.coalesce(
            F.col('value_to_convert') / rate,
            F.when(df1.from_curr == df1.to_curr, df1.value_to_convert)
        ).alias('value_converted')
    )
)
df.show()
#  --------- ------- ---------- ---------------- ------------------ 
# |from_curr|to_curr|      Date|value_to_convert|   value_converted|
#  --------- ------- ---------- ---------------- ------------------ 
# |      AED|    EUR|2017-03-24|            2000| 390.3962521959789|
# |      AED|    EUR|2017-03-27|             189|19.894736842105264|
# |      EUR|    EUR|2020-01-01|              11|              11.0|
# |      DZD|    EUR|2017-01-12|             130|21.241830065359476|
#  --------- ------- ---------- ---------------- ------------------ 

標籤：阿帕奇火花日期pysparkapache-spark-sql最近的

上一篇：CONCAT在同一查詢中新創建的兩個日期和時間欄位

下一篇：了解日期類