如何优化PySpark DataFrame字符串列分词匹配的慢自连接操作

性能根因 #

你现有方案的核心性能问题是同品牌同品类分组下的笛卡尔积冗余计算：
你用tz_brandname、producttype两个字段做等值连接，同组内所有记录会先做全量笛卡尔积，再按分词匹配规则过滤，若单个品牌+品类下有N个产品，会生成N²条临时数据，数据量稍大就会严重卡顿，不管用多条件OR还是array_intersect都绕不开这个问题。

优化方案 #

核心思路是把非等值过滤规则转化为等值连接条件，完全避免笛卡尔积：

1. 预处理A表：仅保留需要的字段，同时计算每条产品的首分词
2. 预处理B表：把每个产品名的前7个分词拆分为独立行，后续直接用分词做等值匹配
3. 用三个等值条件做连接：品牌相等、品类相等、A表首分词 = B表拆分出的分词，Spark会自动按这三个键做shuffle分区，仅匹配到的记录才会被拉到同一节点，计算量会下降几个量级。

优化后代码 #

from pyspark.sql.functions import split, upper, trim, explode, col, broadcast

# 1. 预处理A表：计算首分词
a_df = products_productname_df.select(
    "tz_brandname",
    "producttype",
    col("productname").alias("a_name"),
    upper(trim(split("productname", "\\s+")[0])).alias("first_token")
).distinct()

# 2. 预处理B表：拆分前7个分词并炸开为行
b_df = products_productname_df.select(
    "tz_brandname",
    "producttype",
    col("productname").alias("b_name"),
    # 拆分后取前7个分词，统一大小写和去空格
    split(upper(trim(col("productname"))), "\\s+")[0:7].alias("top7_tokens")
).distinct()
# 炸开分词，过滤空值
b_explode_df = b_df.select(
    "tz_brandname",
    "producttype",
    "b_name",
    explode("top7_tokens").alias("match_token")
).filter(col("match_token").isNotNull())

# 3. 等值连接，避免笛卡尔积
# 如果b_explode_df数据量不大可以加broadcast(b_explode_df)进一步提速
result_df = a_df.join(
    b_explode_df,
    (a_df.tz_brandname == b_explode_df.tz_brandname)
    & (a_df.producttype == b_explode_df.producttype)
    & (a_df.first_token == b_explode_df.match_token)
    & (a_df.a_name != b_explode_df.b_name), # 排除自己匹配自己的情况
    how="inner"
).select(
    a_df.tz_brandname,
    a_df.producttype,
    "a_name",
    "b_name",
    "first_token"
).distinct()

result_df.orderBy("a_name").show(truncate=False)

额外优化建议 #

• 如果单个品牌+品类下的产品量超过10万，可以对预处理后的a_df和b_explode_df按tz_brandname、producttype做分区预存储，避免每次计算都重复预处理
• 分词拆分时可以提前过滤无意义的停用词，减少无效匹配的计算量

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