如何无需Join在PySpark DataFrame中基于Date2映射生成Value2列

针对你的需求，有两种不用Join操作的方法可以实现，前提是同一个Date对应的Value是唯一的（从你的数据来看符合这个条件）：

方案1：广播变量 + UDF #

利用广播变量传递Date到Value的映射字典，通过UDF实现快速查找：

步骤与代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import DoubleType

# 初始化Spark会话
spark = SparkSession.builder.appName("get_value2").getOrCreate()

# 构造原始DataFrame
data = [
    ("2019/01/10", 9.5, None),
    ("2019/01/10", 9.5, None),
    ("2019/01/11", 4.5, "2019/01/10"),
    ("2019/01/12", 6.7, "2019/01/11"),
    ("2019/01/12", 6.7, "2019/01/10"),
    ("2019/01/13", 9.2, "2019/01/12"),
    ("2019/01/14", 13.6, "2019/01/13"),
    ("2019/01/15", 2.7, "2019/01/14"),
    ("2019/01/16", 7.8, "2019/01/15")
]
df = spark.createDataFrame(data, ["Date", "Value", "Date2"])

# 提取Date与Value的唯一映射字典
date_value_map = df.select("Date", "Value").distinct().rdd.collectAsMap()
# 广播映射字典（避免节点重复传输数据）
broadcast_map = spark.sparkContext.broadcast(date_value_map)

# 定义UDF：根据Date2查找对应的Value
def get_value2(date2):
    return broadcast_map.value.get(date2) if date2 else None

get_value2_udf = udf(get_value2, DoubleType())

# 生成Value2列
result_df = df.withColumn("Value2", get_value2_udf(df["Date2"]))

# 查看结果
result_df.show()

方案2：窗口函数 + 映射查找 #

通过窗口函数收集全局的Date-Value映射，再用UDF逐行匹配：

步骤与代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, collect_set, struct
from pyspark.sql.types import DoubleType, StringType

spark = SparkSession.builder.appName("window_value_match").getOrCreate()

# 构造原始DataFrame（同方案1）
data = [
    ("2019/01/10", 9.5, None),
    ("2019/01/10", 9.5, None),
    ("2019/01/11", 4.5, "2019/01/10"),
    ("2019/01/12", 6.7, "2019/01/11"),
    ("2019/01/12", 6.7, "2019/01/10"),
    ("2019/01/13", 9.2, "2019/01/12"),
    ("2019/01/14", 13.6, "2019/01/13"),
    ("2019/01/15", 2.7, "2019/01/14"),
    ("2019/01/16", 7.8, "2019/01/15")
]
df = spark.createDataFrame(data, ["Date", "Value", "Date2"])

# 收集全局的Date-Value映射集合
df_with_map = df.withColumn(
    "date_value_list",
    collect_set(struct("Date", "Value")).over()
)

# 定义UDF：从映射集合中匹配Date2对应的Value
def match_value(date2, map_list):
    if not date2:
        return None
    for item in map_list:
        if item["Date"] == date2:
            return item["Value"]
    return None

match_value_udf = udf(match_value, DoubleType())

# 生成Value2列并删除临时列
result_df = df_with_map.withColumn(
    "Value2",
    match_value_udf(df_with_map["Date2"], df_with_map["date_value_list"])
).drop("date_value_list")

# 查看结果
result_df.show()

注意事项 #

• 如果同一Date存在多个不同的Value，需要先对每个Date的Value做聚合（比如取平均值、最大值），再生成映射关系。
• 方案1的广播变量更适合大数据场景，传输效率更高；方案2的窗口函数会在每行附加映射集合，数据量较大时内存占用更高。

内容的提问来源于stack exchange，提问作者Javier