Spark v2.3中pyspark.ml.fpm(FP Growth)运行缓慢问题咨询

绝对，数据倾斜大概率就是你遇到Spark FP Growth后期任务拖慢的核心诱因——我在Spark 2.x环境里处理关联规则挖掘时也踩过一模一样的坑。

为什么数据倾斜会导致FP Growth后期变慢？ #

FP Growth算法分为两个核心阶段：

1. 生成频繁项集：这个阶段通常是分布式处理的，数据分散在各个节点，速度相对稳定；
2. 生成关联规则：这个阶段需要基于频繁项集做笛卡尔积运算，生成所有可能的规则候选。如果数据中存在高频项（比如某个商品出现在80%以上的交易记录里），那么所有包含这个项的频繁项集都会被分配到同一个Shuffle分区，导致少数几个Executor承担了绝大多数的计算量，直接拖慢整个任务的进度——这就是Spark UI里看到的“后期速度极慢”的典型表现。

不修改阈值/删列的可行解决方案 #

既然你不愿意调整minSupport/minConfidence，也不能删除列，以下几个方案亲测在Spark 2.3环境中有效：

1. 自定义分区策略打散高频项

核心思路是给高频项添加随机后缀，让原本集中在一个分区的数据分散到多个分区处理，最后再还原结果：

from pyspark.sql import functions as F
import random

# 第一步：找出高频项（可根据数据量调整阈值，比如取出现次数前5%的项）
freq_itemsets = model.freqItemsets
item_freq = freq_itemsets.select(F.explode("items").alias("item"), "freq") \
    .groupBy("item").agg(F.sum("freq").alias("total_freq"))
high_freq_threshold = item_freq.approxQuantile("total_freq", [0.95], 0.01)[0]
high_freq_items = item_freq.filter(F.col("total_freq") > high_freq_threshold) \
    .select("item").rdd.flatMap(lambda x: x).collect()
high_freq_broadcast = sc.broadcast(high_freq_items)

# 第二步：给包含高频项的规则候选添加随机后缀，打散分区
def add_random_suffix(row):
    antecedent, consequent = row.antecedent, row.consequent
    # 判断是否包含高频项
    has_high_freq = any(item in high_freq_broadcast.value for item in antecedent + consequent)
    if has_high_freq:
        suffix = str(random.randint(1, 10))  # 生成1-10的随机后缀，可根据集群规模调整
        new_antecedent = tuple([f"{item}_{suffix}" for item in antecedent])
        new_consequent = tuple([f"{item}_{suffix}" for item in consequent])
        return (new_antecedent, new_consequent, row.confidence, row.lift)
    else:
        return (antecedent, consequent, row.confidence, row.lift)

# 转换RDD并重新分区
rules_rdd = model.associationRules.rdd.map(add_random_suffix).repartition(500)  # 分区数根据数据量调整

# 第三步：还原后缀并去重
def remove_suffix(row):
    antecedent = tuple([item.split("_")[0] for item in row[0]])
    consequent = tuple([item.split("_")[0] for item in row[1]])
    return (antecedent, consequent, row[2], row[3])

final_rules = rules_rdd.map(remove_suffix).distinct().toDF(["antecedent", "consequent", "confidence", "lift"])

2. 调整Spark Shuffle相关参数

通过优化Shuffle配置，减轻倾斜带来的压力：

• 增大spark.sql.shuffle.partitions：默认是200，可根据数据量调整到1000-2000，让Shuffle数据更均匀分布；
• 开启外部Shuffle服务：减少Executor的Shuffle开销；
• 调整合并阈值：避免小文件过多导致的IO瓶颈。

提交任务时可以这样设置：

spark-submit \
    --conf spark.sql.shuffle.partitions=1500 \
    --conf spark.shuffle.service.enabled=true \
    --conf spark.shuffle.sort.bypassMergeThreshold=1000 \
    --conf spark.executor.memory=8g \
    your_fp_growth_script.py

3. 提前过滤无效规则候选

基于频繁项集的频率，提前过滤掉不可能满足minConfidence的规则组合，减少后续计算量：

# 先把频繁项集转换成字典，方便快速查询频率
freq_dict = freq_itemsets.rdd.map(lambda row: (tuple(row.items), row.freq)).collectAsMap()
freq_dict_broadcast = sc.broadcast(freq_dict)

def filter_invalid_candidates(row):
    antecedent = tuple(row.antecedent)
    combined = tuple(sorted(antecedent + row.consequent))
    # 计算置信度预估值：freq(combined)/freq(antecedent)
    if freq_dict_broadcast.value.get(antecedent, 0) == 0:
        return False
    estimated_conf = freq_dict_broadcast.value.get(combined, 0) / freq_dict_broadcast.value[antecedent]
    # 只保留预估值大于等于minConfidence的候选
    return estimated_conf >= model._java_obj.getMinConfidence()

# 过滤后再处理
filtered_rules = model.associationRules.filter(filter_invalid_candidates)

这些方案都不需要修改你的核心阈值参数，也不用删除任何列，能有效缓解数据倾斜带来的任务卡顿问题。

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