Spark按tz_brandname与producttype分组生成productname全配对的高效方案

有，比自连接性能更好的实现方式是分组聚合+自定义函数生成配对+爆炸展开，该方案仅需要一次shuffle操作，对比需要两次shuffle的自连接实现，常规场景下性能可提升1倍以上。

实现逻辑 #

• 按tz_brandname、producttype两个字段分组，将同组所有productname聚合为列表
• 自定义UDF遍历列表生成所有有序两两配对（排除自身与自身配对的情况，匹配预期输出逻辑）
• 炸开配对结果，拆解为两列即可得到最终输出

代码示例 #

PySpark 实现 #

from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType, StructType, StructField, StringType

# 定义生成有序两两配对的逻辑
def generate_ordered_pairs(product_list):
    pairs = []
    list_len = len(product_list)
    for i in range(list_len):
        for j in range(list_len):
            if i != j:
                pairs.append((product_list[i], product_list[j]))
    return pairs

# 注册UDF
pair_schema = ArrayType(StructType([
    StructField("product1", StringType(), nullable=False),
    StructField("product2", StringType(), nullable=False)
]))
pair_udf = F.udf(generate_ordered_pairs, pair_schema)

# 核心处理逻辑
result_df = df \
    .groupBy("tz_brandname", "producttype") \
    .agg(F.collect_list("productname").alias("product_list")) \
    .withColumn("pair", F.explode(pair_udf("product_list"))) \
    .select(
        "tz_brandname",
        "producttype",
        F.col("pair.product1").alias("productname"),
        F.col("pair.product2").alias("productname")
    )

大分组场景适配 #

如果存在单分组内productname数量超过1000的大组，为避免单Task内存压力可以使用加盐优化的自连接，性能依然优于普通自连接：

# 加盐自连接示例
salt_count = 10 # 可根据总数据量调整盐值数量
df_salted = df.withColumn("salt", F.floor(F.rand() * salt_count))

df_left = df_salted.selectExpr(
    "tz_brandname", "producttype", "productname as productname_left", "salt"
)
df_right = df_salted.selectExpr(
    "tz_brandname as tz_brandname_r", 
    "producttype as producttype_r", 
    "productname as productname_right", 
    "salt as salt_r"
)

result_df = df_left.join(
    df_right,
    (df_left.tz_brandname == df_right.tz_brandname_r) &
    (df_left.producttype == df_right.producttype_r) &
    (df_left.salt == df_right.salt_r) &
    (df_left.productname_left != df_right.productname_right)
).select(
    "tz_brandname", "producttype", 
    F.col("productname_left").alias("productname"),
    F.col("productname_right").alias("productname")
).distinct()

性能对比 #

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