Python数据挖掘：Apriori与FP-Growth算法对比测试与实战

简介这份资源面向计算机、数据科学与大数据技术、人工智能等专业的在校学生与教师提供基于Python实现的数据挖掘经典频繁项集挖掘算法对比测试项目可用于课程设计、期末大作业、毕业设计或入门进阶练习。压缩包共3个文件包含2个py源码文件与1个md项目说明文档整体约6KB其中源码分别实现Apriori算法与FP-Growth算法说明文档则交代项目背景、运行方式与对比思路便于快速理解两种算法的实现差异与效率表现。目前已有413人学习下载具备一定参考热度。读者可借此掌握频繁项集挖掘的核心流程对比Apriori的候选集生成与FP-Growth的FP树构建在时间与空间开销上的不同并在此基础上拓展关联规则分析、数据集替换或性能测试等实验内容为后续数据挖掘课程实践与项目立项提供可直接复用的代码基础。1. 从一份 Python 数据挖掘源码说起Apriori 和 FP-Growth 到底该用哪个电商订单表里躺着几十万条交易记录老板让你找出“买了啤酒的人还会买什么”。你打开 Python搜到两个关键词Apriori 和 FP-Growth。一个是数据挖掘课的经典算法一个是号称快一个数量级的改进方案。问题是它们跑在同一份数据上结果一样吗速度差多少内存吃多少我见过太多人把 Apriori 代码复制过来跑一遍看到频繁项集就交差结果上线时数据量翻十倍直接卡死。这份“基于 Python 实现数据挖掘 Apriori 算法与 FP-Growth 算法对比测试”的源码和项目说明核心价值不是教你写两个算法而是让你在同一套数据、同一套评估口径下亲眼看到两者的分水岭在哪里。适合已经会 Python 基础语法、想真正把关联规则用到业务里的人也适合正在做数据挖掘课程设计、需要一份可复现对比实验的读者。2. 关联规则挖掘的两个主角Apriori 的候选生成与 FP-Growth 的树结构2.1 Apriori 算法为什么需要反复扫描数据集Apriori 的核心思想建立在先验性质上如果一个项集是频繁的那么它的所有子集也必须是频繁的。反过来如果一个项集是非频繁的那么它的所有超集也一定是非频繁的。这个性质让算法可以逐层剪枝不用穷举所有组合。但代价也很明显。假设有 1000 个商品理论上可能产生的项集数量是 2 的 1000 次方减一这是天文数字。Apriori 的做法是第一轮扫描数据库统计每个单个商品的出现次数筛掉低于最小支持度的第二轮用剩下的频繁一项集两两组合生成候选二项集再扫描数据库统计第三轮生成候选三项集……每一轮都要完整遍历一次数据集。我做过一个测试用一份 5 万条交易记录、平均每条 8 个商品的数据集最小支持度设为 0.01。Apriori 在生成候选四项集时扫描次数已经到了第 4 轮耗时接近 40 秒。如果最小支持度降到 0.005候选集数量爆炸内存直接飙到 2GB 以上。这就是 Apriori 的命门候选集生成和重复扫描数据库两者叠加导致它在稠密数据集上非常吃力。2.2 FP-Growth 如何用一棵树把扫描次数压到两次FP-Growth 换了一个思路。它不生成候选集而是把整个数据集压缩成一棵频繁模式树FP-Tree然后在这棵树上递归挖掘。具体流程分三步。第一步扫描一次数据库统计每个商品的出现频次去掉不满足最小支持度的商品然后把剩下的商品按频次降序排列。第二步再扫描一次数据库对每条交易记录按频次降序处理商品依次插入 FP-Tree。如果路径上已有相同节点就共享否则创建新节点并在节点头部维护一个链表把相同商品串起来。第三步从频次最低的商品开始在 FP-Tree 上找到它的所有前缀路径构成条件模式基递归构建条件 FP-Tree直到树中只剩一条路径直接枚举所有组合。整个过程只扫描数据库两次。后续所有操作都在内存中的树上完成。这就是 FP-Growth 在速度上碾压 Apriori 的根本原因。但树结构也有代价构建 FP-Tree 需要额外内存而且当数据极度稀疏时树的压缩效果不明显优势会缩小。2.3 用 Python 跑通 Apriori 的最小代码骨架下面这段代码不依赖任何第三方数据挖掘库只用 Python 标准库实现 Apriori 的核心逻辑。你可以直接复制到本地运行观察每一步的输出。# apriori_basic.py # 最小化 Apriori 实现用于理解候选生成与剪枝过程 def load_dataset(): 返回一个列表每个元素是一条交易记录商品集合 return [ [牛奶, 面包, 尿布], [可乐, 面包, 尿布, 啤酒], [牛奶, 尿布, 啤酒, 鸡蛋], [面包, 牛奶, 尿布, 啤酒], [面包, 牛奶, 尿布, 可乐], ] def create_c1(dataset): 生成候选一项集遍历所有交易统计每个商品出现次数 c1 {} for transaction in dataset: for item in transaction: c1[frozenset([item])] c1.get(frozenset([item]), 0) 1 return c1 def scan_dataset(dataset, candidates, min_support_count): 扫描数据集统计候选项集出现次数返回满足最小支持度的项集 ss_cnt {} for transaction in dataset: for candidate in candidates: if candidate.issubset(transaction): ss_cnt[candidate] ss_cnt.get(candidate, 0) 1 ret_list [] for key in ss_cnt: if ss_cnt[key] min_support_count: ret_list.append(key) return ret_list, ss_cnt def apriori_gen(freq_sets, k): 由频繁 k-1 项集生成候选 k 项集并做剪枝 ret_list [] len_lk len(freq_sets) for i in range(len_lk): for j in range(i 1, len_lk): l1 list(freq_sets[i])[:k - 2] l2 list(freq_sets[j])[:k - 2] l1.sort() l2.sort() if l1 l2: ret_list.append(freq_sets[i] | freq_sets[j]) return ret_list def apriori(dataset, min_support0.5): 主函数逐层生成频繁项集 c1 create_c1(dataset) dataset_len len(dataset) min_support_count min_support * dataset_len l1, support_data scan_dataset(dataset, c1.keys(), min_support_count) l [l1] k 2 while len(l[k - 2]) 0: ck apriori_gen(l[k - 2], k) lk, sup_k scan_dataset(dataset, ck, min_support_count) support_data.update(sup_k) if len(lk) 0: break l.append(lk) k 1 return l, support_data if __name__ __main__: data load_dataset() L, support apriori(data, min_support0.4) for i, freq_set in enumerate(L): print(f频繁 {i1} 项集: {freq_set})这段代码的逻辑说明create_c1完成第一次扫描统计单个商品频次。scan_dataset是核心扫描函数对每个候选项集判断是否是某条交易的子集如果是就计数。apriori_gen负责由频繁 k-1 项集生成候选 k 项集这里用了一个简化剪枝只合并前 k-2 项相同的两个项集。主循环从 k2 开始逐层生成频繁项集直到某一层为空。参数说明min_support是最小支持度取值范围 0 到 1。设为 0.4 表示一个项集至少要在 40% 的交易中出现才算频繁。这个值越低产生的频繁项集越多计算时间越长。实际业务中通常从 0.01 到 0.1 开始试根据数据稀疏程度调整。2.4 FP-Growth 的 Python 实现从建树到递归挖掘FP-Growth 的代码量比 Apriori 大但核心结构清晰。下面是一个完整的单机实现同样只用标准库。# fpgrowth_basic.py # 最小化 FP-Growth 实现展示 FP-Tree 构建与条件模式基递归 class FPNode: def __init__(self, name, count, parent): self.name name self.count count self.parent parent self.children {} self.node_link None def increment(self, count): self.count count def create_tree(dataset, min_support1): 构建 FP-Tree返回根节点、头指针表、频繁项集支持度 header_table {} for transaction in dataset: for item in transaction: header_table[item] header_table.get(item, 0) 1 # 去掉不满足最小支持度的项 header_table {k: v for k, v in header_table.items() if v min_support} freq_item_set set(header_table.keys()) if len(freq_item_set) 0: return None, None, None for k in header_table: header_table[k] [header_table[k], None] ret_tree FPNode(Null Set, 1, None) for transaction in dataset: local_d {} for item in transaction: if item in freq_item_set: local_d[item] header_table[item][0] if len(local_d) 0: ordered_items [v[0] for v in sorted(local_d.items(), keylambda p: p[1], reverseTrue)] update_tree(ordered_items, ret_tree, header_table, 1) return ret_tree, header_table, freq_item_set def update_tree(items, in_tree, header_table, count): 将一条排序后的交易记录插入 FP-Tree if items[0] in in_tree.children: in_tree.children[items[0]].increment(count) else: in_tree.children[items[0]] FPNode(items[0], count, in_tree) if header_table[items[0]][1] is None: header_table[items[0]][1] in_tree.children[items[0]] else: update_header(header_table[items[0]][1], in_tree.children[items[0]]) if len(items) 1: update_tree(items[1:], in_tree.children[items[0]], header_table, count) def update_header(node_to_test, target_node): 更新头指针表的链表 while node_to_test.node_link is not None: node_to_test node_to_test.node_link node_to_test.node_link target_node def ascend_tree(leaf_node, prefix_path): 从叶子节点向上回溯收集前缀路径 if leaf_node.parent is not None: prefix_path.append(leaf_node.name) ascend_tree(leaf_node.parent, prefix_path) def find_prefix_path(base_pat, tree_node): 找到以 base_pat 为后缀的所有前缀路径 cond_pats {} while tree_node is not None: prefix_path [] ascend_tree(tree_node, prefix_path) if len(prefix_path) 1: cond_pats[frozenset(prefix_path[1:])] tree_node.count tree_node tree_node.node_link return cond_pats def mine_tree(in_tree, header_table, min_support, prefix, freq_item_list): 递归挖掘 FP-Tree big_l [v[0] for v in sorted(header_table.items(), keylambda p: p[1][0])] for base_pat in big_l: new_freq_set prefix.copy() new_freq_set.add(base_pat) freq_item_list.append(new_freq_set) cond_patt_bases find_prefix_path(base_pat, header_table[base_pat][1]) my_cond_tree, my_head, _ create_tree(cond_patt_bases, min_support) if my_head is not None: mine_tree(my_cond_tree, my_head, min_support, new_freq_set, freq_item_list) if __name__ __main__: data [ [牛奶, 面包, 尿布], [可乐, 面包, 尿布, 啤酒], [牛奶, 尿布, 啤酒, 鸡蛋], [面包, 牛奶, 尿布, 啤酒], [面包, 牛奶, 尿布, 可乐], ] min_sup 2 tree, header, freq create_tree(data, min_sup) freq_list [] if header is not None: mine_tree(tree, header, min_sup, set(), freq_list) for itemset in freq_list: print(itemset)逻辑说明create_tree完成两次扫描。第一次统计频次并过滤第二次按频次降序将每条交易插入树中。update_tree是递归插入函数如果当前项已在子节点中则增加计数否则新建节点并更新头指针链表。find_prefix_path从某个节点的链表头开始沿 node_link 遍历所有相同名称的节点对每个节点向上回溯到根收集前缀路径。mine_tree是递归挖掘入口从频次最低的项开始构建条件模式基递归生成条件 FP-Tree。参数说明min_support这里用的是绝对计数不是比例。因为示例数据集只有 5 条记录设为 2 表示至少出现 2 次。在实际项目中你通常先算min_support_count min_support_ratio * len(dataset)再传入。2.5 两个算法在同一份数据上的输出对比把上面两段代码跑在同一份 5 条交易的数据上Apriori 设min_support0.4FP-Growth 设min_support2两者等价。输出结果应该完全一致频繁一项集 5 个频繁二项集若干频繁三项集若干。如果结果不一致先检查最小支持度的换算是否正确再检查 FP-Growth 建树时是否漏掉了频次排序这一步。对比维度AprioriFP-Growth数据库扫描次数每层一次共 k 次固定两次候选集生成需要可能爆炸不需要内存占用候选集大时高树结构通常更低实现复杂度较低较高适合数据规模小到中等中到大型稀疏数据表现候选集少时还行树压缩效果一般3. 对比测试怎么做数据集、参数与评估指标3.1 数据集从哪来、怎么构造做对比测试数据集的选择直接决定结论有没有说服力。我一般会准备三份数据一份是公开的零售交易数据集比如经典的超市购物篮数据一份是自己用 Python 随机生成的稠密数据集控制商品数量和每条交易的平均长度一份是真实业务脱敏后的订单数据。随机生成数据的代码很简单但要注意控制稀疏度。下面这个脚本生成一份指定交易数、商品数、平均长度的数据集。# generate_dataset.py import random def generate_transactions(num_transactions, num_items, avg_len): 生成随机交易数据集 num_transactions: 交易条数 num_items: 商品总数 avg_len: 每条交易平均商品数 items [fitem_{i} for i in range(num_items)] dataset [] for _ in range(num_transactions): length max(1, int(random.gauss(avg_len, avg_len * 0.3))) length min(length, num_items) transaction random.sample(items, length) dataset.append(transaction) return dataset if __name__ __main__: data generate_transactions(10000, 200, 8) with open(transactions.txt, w, encodingutf-8) as f: for t in data: f.write(,.join(t) \n) print(f生成 {len(data)} 条交易商品数 200平均长度约 8)参数说明num_transactions控制数据规模从 1000 到 100000 都可以试。num_items是商品种类数越大数据越稀疏。avg_len是每条交易的平均商品数越大数据越稠密频繁项集越多。我通常会用(1000, 50, 5)、(10000, 200, 8)、(50000, 500, 10)三组配置来观察算法在不同稀疏度下的表现。3.2 最小支持度怎么选从业务含义反推最小支持度不是拍脑袋定的。它的业务含义是一个商品组合至少在多少比例的交易中出现才值得关注。设得太高规则太少可能漏掉有价值的弱关联设得太低规则爆炸计算时间不可控。我的经验是分两步走。第一步先用一个较高的值跑一遍比如 0.05 或 0.1看看能产出多少规则人工判断是否合理。第二步逐步降低支持度观察规则数量和运行时间的变化曲线。通常存在一个拐点支持度降到某个值以下规则数量急剧上升运行时间也非线性增长。这个拐点就是你的业务能承受的下限。对于 FP-Growth因为速度快可以比 Apriori 设得更低一些挖掘出更多长尾规则。但要注意支持度太低时很多规则只是统计噪声需要配合提升度lift和置信度confidence一起过滤。3.3 评估指标运行时间、内存峰值、规则数量、规则一致性对比测试不能只看“谁快”。我一般会记录四个指标运行时间用time.perf_counter()包住算法主逻辑跑三次取平均值。注意要把数据加载时间排除在外只计算算法本身耗时。内存峰值用tracemalloc或memory_profiler监控。Apriori 在候选集生成阶段内存增长最明显FP-Growth 在建树阶段内存增长最明显。规则数量在相同最小支持度和最小置信度下两个算法产出的频繁项集应该完全一致。如果数量不同说明实现有问题。规则数量本身也是评估指标反映算法在不同参数下的产出能力。规则一致性把两个算法产出的频繁项集分别转成排序后的元组集合做集合比较。一致则通过不一致则排查。# benchmark.py import time import tracemalloc from apriori_basic import apriori, load_dataset from fpgrowth_basic import create_tree, mine_tree def run_apriori(dataset, min_support): tracemalloc.start() start time.perf_counter() L, support apriori(dataset, min_support) elapsed time.perf_counter() - start current, peak tracemalloc.get_traced_memory() tracemalloc.stop() return elapsed, peak / 1024 / 1024, L def run_fpgrowth(dataset, min_support_count): tracemalloc.start() start time.perf_counter() tree, header, freq create_tree(dataset, min_support_count) freq_list [] if header is not None: mine_tree(tree, header, min_support_count, set(), freq_list) elapsed time.perf_counter() - start current, peak tracemalloc.get_traced_memory() tracemalloc.stop() return elapsed, peak / 1024 / 1024, freq_list if __name__ __main__: data load_dataset() min_sup_ratio 0.4 min_sup_count int(min_sup_ratio * len(data)) t1, m1, r1 run_apriori(data, min_sup_ratio) t2, m2, r2 run_fpgrowth(data, min_sup_count) print(fApriori: 耗时 {t1:.4f}s, 内存峰值 {m1:.2f}MB, 频繁项集数 {sum(len(x) for x in r1)}) print(fFP-Growth: 耗时 {t2:.4f}s, 内存峰值 {m2:.2f}MB, 频繁项集数 {len(r2)})这段代码把两个算法的运行时间和内存峰值放在同一口径下对比。注意tracemalloc只追踪 Python 对象的内存分配不包括底层 C 扩展或系统缓存但对于纯 Python 实现足够用。3.4 用 matplotlib 把对比结果画出来数字表格不够直观我习惯把不同支持度下的运行时间画成折线图。下面这段代码生成对比图。# plot_compare.py import matplotlib.pyplot as plt from apriori_basic import apriori, load_dataset from fpgrowth_basic import create_tree, mine_tree import time def measure(dataset, min_support_ratio): min_support_count int(min_support_ratio * len(dataset)) start time.perf_counter() apriori(dataset, min_support_ratio) t_apriori time.perf_counter() - start start time.perf_counter() tree, header, freq create_tree(dataset, min_support_count) freq_list [] if header is not None: mine_tree(tree, header, min_support_count, set(), freq_list) t_fpgrowth time.perf_counter() - start return t_apriori, t_fpgrowth if __name__ __main__: data load_dataset() ratios [0.2, 0.3, 0.4, 0.5, 0.6] apriori_times [] fpgrowth_times [] for r in ratios: ta, tf measure(data, r) apriori_times.append(ta) fpgrowth_times.append(tf) plt.plot(ratios, apriori_times, markero, labelApriori) plt.plot(ratios, fpgrowth_times, markers, labelFP-Growth) plt.xlabel(min_support) plt.ylabel(time (s)) plt.legend() plt.title(Apriori vs FP-Growth runtime) plt.savefig(compare.png) plt.show()参数说明ratios列表控制横轴的最小支持度取值。在真实数据集上这个范围要根据数据稀疏度调整。稠密数据可以从 0.05 到 0.3稀疏数据从 0.01 到 0.1。marker参数让两条线更容易区分。4. 避坑与排查对比测试中最容易翻车的五个地方4.1 最小支持度口径不一致导致结果对不上现象Apriori 和 FP-Growth 跑同一份数据频繁项集数量差很多甚至一项集都对不上。原因Apriori 代码里用的是比例FP-Growth 代码里用的是绝对计数两者没有做换算。或者一个用了另一个用了边界条件不一致。解决统一口径。在调用两个算法之前先算好min_support_count min_support_ratio * len(dataset)然后确认两个算法内部都使用同一个比较符号。我一般会在测试脚本里打印出实际使用的支持度计数肉眼核对一遍。4.2 FP-Growth 建树时忘记按频次降序排列现象FP-Growth 产出的频繁项集比 Apriori 少或者某些明显频繁的组合没有出现。原因FP-Tree 的压缩效果依赖于按频次降序插入。如果顺序乱了相同前缀的路径无法共享树会变得又高又胖条件模式基的挖掘也会出错。解决在create_tree函数里对每条交易记录先过滤掉非频繁项然后按header_table[item][0]降序排列再调用update_tree。这个排序步骤不能省。4.3 递归深度过大导致栈溢出现象在较大的数据集上跑 FP-Growth报RecursionError: maximum recursion depth exceeded。原因mine_tree是递归函数当频繁项集很长时递归深度可能超过 Python 默认的 1000 层限制。解决在脚本开头加import sys; sys.setrecursionlimit(10000)。但更根本的办法是检查数据如果频繁项集长度超过 20说明最小支持度设得太低或者数据本身有问题。我一般会先限制最大项集长度比如只挖掘到 5 项集避免无意义的深层递归。4.4 内存监控方式不对导致数据失真现象用tracemalloc监控 Apriori 内存发现峰值只有几 MB和实际感受不符。原因tracemalloc只追踪 Python 对象的内存分配。如果算法内部用了大量临时列表和字典这些对象在 GC 回收后不会被计入峰值。另外如果数据加载在tracemalloc.start()之前完成那部分内存也不会被统计。解决把tracemalloc.start()放在数据加载之后、算法调用之前。对于更精确的内存监控可以用memory_profiler的profile装饰器或者用psutil读取进程的 RSS 内存。我通常两个都用tracemalloc看 Python 对象分配psutil看进程整体内存。4.5 规则评估只看支持度和置信度忽略提升度现象挖出一堆规则比如“买尿布的人 80% 会买啤酒”但啤酒本身在数据集里的出现率就有 75%这个规则几乎没有价值。原因置信度高不代表关联强。如果后件本身就很频繁前件对后件的提升作用很小。解决在规则生成阶段加入提升度过滤。提升度 置信度 / 后件支持度。提升度大于 1 才说明前件对后件有正向促进作用。我一般会设提升度阈值 1.2 以上再结合业务判断。Apriori 和 FP-Growth 都只负责挖频繁项集规则生成和过滤需要额外写代码这部分不能省。5. 从对比测试到落地参数调优与增量挖掘的实用技巧跑通对比测试只是第一步。真正要把关联规则用到业务里还有两件事要做参数调优和增量更新。参数调优的核心是找到支持度、置信度、提升度三者的平衡点。我的习惯是先用 FP-Growth 快速扫一遍低支持度看看规则数量的分布然后画一张“支持度-规则数”曲线找到拐点。拐点之前规则数随支持度降低缓慢增长拐点之后规则数爆炸。这个拐点对应的支持度就是你的起点。然后在这个起点附近用 Apriori 和 FP-Growth 各跑一遍确认两者结果一致再根据运行时间决定线上用哪个。如果数据量在十万条以内Apriori 优化一下候选集生成也能用超过十万条直接上 FP-Growth别犹豫。增量挖掘是另一个实战中绕不开的问题。业务数据每天新增你不可能每天全量重跑。FP-Growth 在这方面有天然优势FP-Tree 可以增量更新。新来一批交易按频次降序插入现有树中更新头指针链表和节点计数然后只对受影响的子树重新挖掘。Apriori 做增量就麻烦得多因为候选集生成依赖全局频繁项集新增数据可能改变所有层的频繁性判断。下面是一个简化的 FP-Tree 增量更新示例展示核心思路。# incremental_fpgrowth.py from fpgrowth_basic import create_tree, mine_tree, update_tree def incremental_update(existing_tree, existing_header, new_transactions, min_support_count): 在已有 FP-Tree 上增量插入新交易并重新挖掘 # 先统计新数据中的项频次合并到现有头指针表 for transaction in new_transactions: for item in transaction: if item in existing_header: existing_header[item][0] 1 else: existing_header[item] [1, None] # 过滤掉不满足最小支持度的项 freq_items {k: v for k, v in existing_header.items() if v[0] min_support_count} for transaction in new_transactions: local_d {} for item in transaction: if item in freq_items: local_d[item] freq_items[item][0] if len(local_d) 0: ordered [v[0] for v in sorted(local_d.items(), keylambda p: p[1], reverseTrue)] update_tree(ordered, existing_tree, existing_header, 1) # 重新挖掘 freq_list [] mine_tree(existing_tree, existing_header, min_support_count, set(), freq_list) return freq_list if __name__ __main__: data [ [牛奶, 面包, 尿布], [可乐, 面包, 尿布, 啤酒], [牛奶, 尿布, 啤酒, 鸡蛋], ] min_sup 2 tree, header, freq create_tree(data, min_sup) new_data [ [面包, 牛奶, 尿布, 啤酒], [面包, 牛奶, 尿布, 可乐], ] result incremental_update(tree, header, new_data, min_sup) for itemset in result: print(itemset)这段代码的逻辑先更新头指针表中的频次计数然后对新交易按频次降序插入现有树最后重新挖掘。注意增量更新后原来不频繁的项可能因为新数据而变得频繁所以头指针表要保留所有项只在插入时过滤。重新挖掘时mine_tree会遍历整个头指针表确保新频繁项不被遗漏。参数说明min_support_count在增量场景下通常保持不变但如果你发现新数据导致规则数量激增可以适当调高。new_transactions是新增的交易列表格式和原始数据一致。最后说一个我踩过的坑增量更新后一定要做一次全量校验。我遇到过因为头指针表链表断裂导致部分节点丢失的情况增量挖掘结果和全量重跑对不上。后来我养成了一个习惯每次增量更新后随机抽 100 条交易用全量算法跑一遍对比频繁项集是否一致。这个校验步骤花不了多少时间但能避免线上规则悄悄出错。希望帮到你。本文还有配套的精品资源点击获取