Azure Databricks：如何高效关联日历表与库存表并补全缺失记录？

我有一个包含month、year两列的DataFrame df_calendar，另有一个包含month、year、plant、stock四列的DataFrame df_stock。假设df_calendar中2022年有12条记录，需核对df_stock中对应每个月份的库存数据。注意df_stock仅存有库存可用月份的记录，若某月份无库存记录，需显示对应plant名称并将stock设为0。df_stock数据量较大，请问应使用哪种Join？若不使用Join，最高效的实现方式是什么？

示例代码 #

calendar_columns = ["month_number", "year_number"]

calendar = [(1, 2022), \
              (2, 2022), \
              (3, 2022)]

df_calendar = spark.createDataFrame(data=calendar, schema = calendar_columns)

stock_columns = ["month_number", "year_number", "plant_id", "stock"]

stock = [(1, 2022, "P1", 10), \
            (2, 2022, "P1", 20), \
            (3, 2022, "P1", 30), \
            (1, 2022, "P2", 15)]

df_stock = spark.createDataFrame(data=stock, schema = stock_columns)

预期输出 #

1. 推荐使用的Join类型 #

应该用交叉连接（Cross Join）生成全量组合，再配合**左外连接（Left Outer Join）**关联库存数据，最后填充空值为0，具体步骤：

• 先从df_stock提取所有唯一的plant，和df_calendar做交叉连接，得到每个工厂对应所有月份的全量组合；
• 将全量组合与df_stock按month、year、plant做左外连接；
• 用coalesce函数把空的stock字段替换为0。

示例代码：

from pyspark.sql.functions import col, lit, coalesce

# 提取唯一工厂列表
df_plants = df_stock.select("plant_id").distinct().withColumnRenamed("plant_id", "Plant")

# 生成工厂+日历的全量组合，并重命名列匹配预期输出
df_full_combo = df_plants.crossJoin(df_calendar)\
    .withColumnRenamed("month_number", "Month")\
    .withColumnRenamed("year_number", "Year")

# 左连接库存数据并填充0
result = df_full_combo.join(
    df_stock.withColumnRenamed("plant_id", "Plant")\
            .withColumnRenamed("month_number", "Month")\
            .withColumnRenamed("year_number", "Year")\
            .withColumnRenamed("stock", "Stock"),
    on=["Plant", "Month", "Year"],
    how="left"
).withColumn("Stock", coalesce(col("Stock"), lit(0)))

result.select("Plant", "Month", "Year", "Stock").show()

这种方式的优势是：交叉连接的对象是小体量的工厂列表和日历表，不会产生过大的中间数据，后续左外连接Spark能自动做优化，适配df_stock数据量大的场景。

2. 不使用Join的高效实现方式 #

可以采用分组+集合匹配的方式，核心是按工厂分组后生成全量日历组合，再匹配已有库存数据，避免显式Join操作：

• 先收集df_calendar的所有月份和年份，转成结构化数组；
• 对df_stock按plant分组，聚合该工厂的所有库存记录；
• 用explode展开全量日历组合，再通过内置函数匹配库存记录，无匹配则赋值为0。

示例代码：

from pyspark.sql.functions import collect_list, struct, explode, expr, lit, coalesce

# 收集日历数据并转成struct数组
calendar_records = df_calendar.collect()
calendar_struct = array(*[
    struct(lit(m).alias("Month"), lit(y).alias("Year")) 
    for m, y in [(row.month_number, row.year_number) for row in calendar_records]
])

# 分组聚合+匹配库存数据
result = df_stock.groupBy("plant_id").agg(
    collect_list(struct(
        col("month_number").alias("Month"),
        col("year_number").alias("Year"),
        col("stock").alias("Stock")
    )).alias("stock_list")
).withColumn("calendar", explode(calendar_struct))\
.select(
    col("plant_id").alias("Plant"),
    col("calendar.Month").alias("Month"),
    col("calendar.Year").alias("Year"),
    coalesce(
        expr("filter(stock_list, x -> x.Month = calendar.Month and x.Year = calendar.Year)[0].Stock"),
        lit(0)
    ).alias("Stock")
)

result.show()

这种方式避免了多字段Join带来的Shuffle开销，仅基于plant字段做分组Shuffle，在df_stock数据量极大时，能有效减少数据传输成本，性能更优。

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