基于PySpark实现按日期匹配汇率的数据转换需求

需求明确 #

现有两张Spark DataFrame：

• df1（交易数据表）：包含id、from_curr、to_curr、Date、item_number、value_to_convert字段
• df2（汇率表）：包含from_curr、To_curr、Date、rate_exchange字段

需要实现：

• 匹配同货币对+对应日期的汇率，将value_to_convert转换为目标货币金额，生成带converted_value字段的df3
• 处理同货币不转换的场景（from_curr与to_curr相同时直接保留原金额）
• 处理单货币多汇率的场景（同一日期、同一货币对存在多条汇率记录）

具体实现步骤 #

1. 预处理汇率表：解决多汇率问题 #

针对同一from_curr、To_curr、Date下的多条汇率记录，提供两种常见业务处理方式，按需选择：

方式1：保留最新汇率（需汇率表含更新时间戳）

如果汇率表有update_time字段，按货币对+日期分组后取最新的一条汇率：

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

window_spec = Window.partitionBy("from_curr", "To_curr", "Date").orderBy(F.desc("update_time"))
cleaned_currency_df = df2.withColumn("rank", F.row_number().over(window_spec)) \
    .filter(F.col("rank") == 1) \
    .drop("rank", "update_time")

方式2：取汇率平均值

若业务允许用平均汇率计算：

cleaned_currency_df = df2.groupBy("from_curr", "To_curr", "Date") \
    .agg(F.avg("rate_exchange").alias("rate_exchange"))

2. 关联交易表与清洗后的汇率表 #

先统一字段大小写避免匹配错误，再通过from_curr、to_curr、Date关联两张表：

# 统一to_curr字段名与汇率表的To_curr一致
df1 = df1.withColumnRenamed("to_curr", "To_curr")
joined_df = df1.join(cleaned_currency_df, 
                     on=["from_curr", "To_curr", "Date"], 
                     how="left")

3. 计算转换后金额，处理同货币场景 #

用when函数分支处理：同货币直接保留原金额，有汇率则计算转换值，无汇率则设为null（可按需调整）：

df3 = joined_df.withColumn(
    "converted_value",
    F.when(F.col("from_curr") == F.col("To_curr"), F.col("value_to_convert"))
    .when(F.col("rate_exchange").isNotNull(), F.col("value_to_convert") * F.col("rate_exchange"))
    .otherwise(F.lit(None))  # 汇率缺失时设为null，也可改为默认值或抛出提示
)

# 恢复原to_curr字段名
df3 = df3.withColumnRenamed("To_curr", "to_curr")

# 保留原表所有字段+转换后金额
df3 = df3.select("id", "from_curr", "to_curr", "Date", "item_number", "value_to_convert", "converted_value")

4. 封装为可复用函数 #

将上述逻辑封装成函数，方便重复调用：

def convert_currency(df_transactions, df_rates, rate_strategy="latest"):
    """
    PySpark 货币转换函数
    :param df_transactions: 交易数据表（对应df1）
    :param df_rates: 汇率表（对应df2）
    :param rate_strategy: 多汇率处理策略，可选"latest"（取最新）或"average"（取平均）
    :return: 包含converted_value的结果DataFrame
    """
    # 统一字段大小写
    df_transactions = df_transactions.withColumnRenamed("to_curr", "To_curr")
    
    # 清洗汇率表
    if rate_strategy == "latest":
        window_spec = Window.partitionBy("from_curr", "To_curr", "Date").orderBy(F.desc("update_time"))
        cleaned_rates = df_rates.withColumn("rank", F.row_number().over(window_spec)) \
            .filter(F.col("rank") == 1) \
            .drop("rank", "update_time")
    elif rate_strategy == "average":
        cleaned_rates = df_rates.groupBy("from_curr", "To_curr", "Date") \
            .agg(F.avg("rate_exchange").alias("rate_exchange"))
    else:
        raise ValueError("rate_strategy must be 'latest' or 'average'")
    
    # 关联表并计算转换金额
    result_df = df_transactions.join(cleaned_rates, on=["from_curr", "To_curr", "Date"], how="left") \
        .withColumn(
            "converted_value",
            F.when(F.col("from_curr") == F.col("To_curr"), F.col("value_to_convert"))
            .when(F.col("rate_exchange").isNotNull(), F.col("value_to_convert") * F.col("rate_exchange"))
            .otherwise(None)
        ).withColumnRenamed("To_curr", "to_curr")
    
    return result_df.select("id", "from_curr", "to_curr", "Date", "item_number", "value_to_convert", "converted_value")

注意事项 #

• 确保两张表的Date字段格式一致（建议用Spark日期类型，而非字符串），否则关联会失败
• 若汇率表无对应货币对+日期的记录，converted_value会设为null，可根据业务需求修改otherwise的逻辑
• 汇率表的To_curr字段大小写需与交易表的to_curr统一，避免匹配失败

内容的提问来源于stack exchange，提问作者f.ivy