Databricks中PySpark并行处理多Delta商店表关联查询的高效方案

在Databricks环境中，我拥有N个存储商店商品信息的Delta表，表结构均包含store、product、sku字段，示例如下：

store_1表 #

store_2表 #

我需要通过sku字段找出所有商店对中相同的商品，当前针对两个商店的查询语句为：

select * from df_store_1 st1 join df_store_2 st2 on st1.sku=st2.sku

该查询返回有效关联结果：

我需要对所有商店对执行此操作，作为PySpark新手，原本计划生成所有商店对列表并循环处理，代码如下：

list_dfs = []

for store1, store2 in list_all_pairs_stores:
   temp_df = spark.sql("select st1.product, st2.product, st1.sku from store1 st1 join store2 st2 on st1.sku=st2.sku").toPandas()
   list_dfs.append(temp_df)

all_equal_products = pd.concat(list_dfs, axis=1)

请问在PySpark中，有什么高效的方式可以并行化这些查询？

你的循环实现存在明显效率问题：每次循环都会单独触发Spark作业，且频繁在Spark与Pandas之间转换数据会带来额外的集群与Driver间数据传输开销。更高效的做法是将所有商店数据合并到单一DataFrame，通过自连接+过滤实现全商店对的关联查询，全程在Spark分布式环境中处理，天然支持并行。

步骤1：合并所有商店表 #

如果表名有规律（如统一前缀store_），可以批量加载并合并：

# 获取所有以store_开头的Delta表
store_tables = [tbl.name for tbl in spark.catalog.listTables() if tbl.name.startswith("store_")]

# 合并所有表到一个DataFrame
all_stores_df = spark.table(store_tables[0])
for tbl_name in store_tables[1:]:
    all_stores_df = all_stores_df.union(spark.table(tbl_name))

步骤2：自连接获取所有唯一商店对的关联商品 #

通过自连接基于sku关联，同时过滤掉商店ID相同的情况，并通过st1.store < st2.store确保商店对唯一（避免重复出现(1,2)和(2,1)）：

from pyspark.sql.functions import col

matched_pairs_df = all_stores_df.alias("st1") \
    .join(all_stores_df.alias("st2"), on="sku") \
    .where(col("st1.store") < col("st2.store")) \
    .select(
        col("st1.store").alias("store_a"),
        col("st2.store").alias("store_b"),
        col("st1.product").alias("product_a"),
        col("st2.product").alias("product_b"),
        col("sku")
    )

# 查看结果
matched_pairs_df.show()

核心优势 #

• 分布式并行处理：所有操作在Spark集群上并行执行，避免循环触发多个小作业的调度开销。
• 无数据转换开销：全程在Spark DataFrame中处理，无需频繁切换到Pandas，减少数据传输成本。
• 自动执行计划优化：Spark会对合并+自连接操作进行分区修剪、Shuffle优化等，比独立查询效率更高。

海量商店场景优化 #

如果商店表数量极大，合并后数据量过高，可以先按sku分组，收集每个sku对应的商店与商品信息，再生成商店对：

from pyspark.sql.functions import collect_list, explode, array
from itertools import combinations
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, ArrayType

# 按sku分组，收集每个sku对应的(store, product)列表
sku_groups_df = all_stores_df.groupBy("sku") \
    .agg(collect_list(array("store", "product")).alias("store_product_list"))

# 定义UDF生成所有不重复的商店对
def generate_pairs(store_product_list):
    pairs = []
    for (s1, p1), (s2, p2) in combinations(store_product_list, 2):
        pairs.append((s1, s2, p1, p2))
    return pairs

pair_schema = ArrayType(StructType([
    StructField("store_a", IntegerType()),
    StructField("store_b", IntegerType()),
    StructField("product_a", StringType()),
    StructField("product_b", StringType())
]))

generate_pairs_udf = udf(generate_pairs, pair_schema)

# 应用UDF并展开结果
sku_pairs_df = sku_groups_df.withColumn("pairs", generate_pairs_udf(col("store_product_list"))) \
    .select("sku", explode("pairs").alias("pair")) \
    .select(
        col("pair.store_a"),
        col("pair.store_b"),
        col("pair.product_a"),
        col("pair.product_b"),
        col("sku")
    )

sku_pairs_df.show()

这种方式先按sku聚合，减少后续配对计算的数据量，适合sku基数较大的场景。

内容的提问来源于stack exchange，提问作者Andrés Bustamante