Spark 关联规则挖掘

关联分析基本概念

关联分析的任务就是从数据集中挖掘出频繁项集，然后从频繁项集中提取出事物之间的强关联规则，辅助决策。

假设上表为某超时用户的购物数据，*TID代表交易流水号，Items代表一次交易的商品。*关联规则中的重要名词定义：

1、事务：每一条交易称为一个事务，例如示例中的数据集就包含四个事务。

2、项：交易的每一个物品称为一个项，例如Cola、Egg等。

3、项集：包含零个或多个项的集合叫做项集，例如{Cola, Egg, Ham}。

4、 k−项集：包含k个项的项集叫做k-项集，例如{Cola}叫做1-项集，{Cola, Egg}叫做2-项集。

5、支持度计数：一个项集出现在几个事务当中，它的支持度计数就是几。例如{Diaper, Beer}出现在事务002、003和004中，所以它的支持度计数是3。

6、支持度：支持度计数除于总的事务数。例如上例中总的事务数为4，{Diaper, Beer}的支持度计数为3，所以它的支持度是3÷4=75%，说明有75%的人同时买了Diaper和Beer。

7、频繁项集：支持度大于或等于某个阈值的项集就叫做频繁项集。例如阈值设为50%时，因为{Diaper, Beer}的支持度是75%，所以它是频繁项集。

8、前件和后件：对于规则{Diaper}→{Beer}，{Diaper}叫做前件，{Beer}叫做后件。

9、置信度：对于规则{Diaper}→{Beer}，{Diaper, Beer}的支持度计数除于{Diaper}的支持度计数，为这个规则的置信度。例如规则{Diaper}→{Beer}的置信度为3÷3=100%。说明买了Diaper的人100%也买了Beer。

10、强关联规则：大于或等于最小支持度阈值和最小置信度阈值的规则叫做强关联规则。关联分析的最终目标就是要找出强关联规则。

得到的强关联规则并不一定为真：此时需要计算规则中样本的相关性系数：即提升度。

11、提升度

Lift(A->B)=Confidence(A->B)/support(B)=support(A,B)/(support(A)support(B)) =lift(B->A) =（AB同时出现的次数X总的订单数）/(包含A的订单数X包含B的订单数)

提升度表示了一条关联规则是否有效，如果提升度大于1，说明规则为有效的强关联规则，提升度小于1，表示规则为无效的强关联规则。如果提升度为1，则表示两事件相互独立。

频繁模式挖掘

了解到上述重要概念以后，关联规则挖掘的重点便在于如何找到频繁模式。以下分别介绍两种频繁模式挖掘的算法。由于本文旨在通过实践的方式，帮助大家使用spark进行关联规则全流程的认知，此外关于频繁模式挖掘算法方面的讲解，也有非常多非常细致的讲解资料，所以本章附上作者认为比较好的讲解文章，便不再赘述

Apriori

https://www.cnblogs.com/pinard/p/6293298.html

FP-growth

https://blog.csdn.net/javastart/article/details/50521453

使用SPARK进行频繁模式挖掘以及关联规则计算

SPARK中有实现好的FPgrowth算法，帮助我们进行频繁模式的挖掘，但是需要我们从挖掘出的频繁模式中，生成规则，并计算规则的置信度与提升度。

首先来看下如何使用spark挖掘出频繁项集：

我们假设有如下的数据：

r z h k p z y x w v u t s s x o n r x z y m t s q e z x z y r q t p

每一行代表一个事务，接下来看spark代码：

from pyspark.mllib.fpm import FPGrowth from pyspark import SparkContext,SparkConf import itertools import os conf=SparkConf().setMaster("local").setAppName("local_test") sc=SparkContext(conf=conf) sc.setLogLevel("ERROR") data = sc.textFile("D:/data_set/asso_data/sample_fpgrowth.txt") NUM_DATA = len(data.collect()) print('总订单数量',NUM_DATA) transactions = data.map(lambda line: line.strip().split(' ')) model = FPGrowth.train(transactions, minSupport=0.3, numPartitions=10) result = model.freqItemsets() tmp = result.collect() for fi in tmp: print(fi) print('频繁项集数量',len(tmp))

我们设定最小支持度为0.3，一共6条数据，那么只要出现过两次及以上的数据便可以看作是频繁项集，挖掘出的结果如下：

FreqItemset(items=['z'], freq=5) FreqItemset(items=['x'], freq=4) FreqItemset(items=['x', 'z'], freq=3) FreqItemset(items=['y'], freq=3) FreqItemset(items=['y', 'x'], freq=3) FreqItemset(items=['y', 'x', 'z'], freq=3) FreqItemset(items=['y', 'z'], freq=3) FreqItemset(items=['r'], freq=3) FreqItemset(items=['r', 'x'], freq=2) FreqItemset(items=['r', 'z'], freq=2) FreqItemset(items=['s'], freq=3) FreqItemset(items=['s', 'y'], freq=2) FreqItemset(items=['s', 'y', 'x'], freq=2) FreqItemset(items=['s', 'y', 'x', 'z'], freq=2) FreqItemset(items=['s', 'y', 'z'], freq=2) FreqItemset(items=['s', 'x'], freq=3) FreqItemset(items=['s', 'x', 'z'], freq=2) FreqItemset(items=['s', 'z'], freq=2) FreqItemset(items=['t'], freq=3) FreqItemset(items=['t', 'y'], freq=3) FreqItemset(items=['t', 'y', 'x'], freq=3) FreqItemset(items=['t', 'y', 'x', 'z'], freq=3) FreqItemset(items=['t', 'y', 'z'], freq=3) FreqItemset(items=['t', 's'], freq=2) FreqItemset(items=['t', 's', 'y'], freq=2) FreqItemset(items=['t', 's', 'y', 'x'], freq=2) FreqItemset(items=['t', 's', 'y', 'x', 'z'], freq=2) FreqItemset(items=['t', 's', 'y', 'z'], freq=2) FreqItemset(items=['t', 's', 'x'], freq=2) FreqItemset(items=['t', 's', 'x', 'z'], freq=2) FreqItemset(items=['t', 's', 'z'], freq=2) FreqItemset(items=['t', 'x'], freq=3) FreqItemset(items=['t', 'x', 'z'], freq=3) FreqItemset(items=['t', 'z'], freq=3) FreqItemset(items=['p'], freq=2) FreqItemset(items=['p', 'r'], freq=2) FreqItemset(items=['p', 'r', 'z'], freq=2) FreqItemset(items=['p', 'z'], freq=2) FreqItemset(items=['q'], freq=2) FreqItemset(items=['q', 'y'], freq=2) FreqItemset(items=['q', 'y', 'x'], freq=2) FreqItemset(items=['q', 'y', 'x', 'z'], freq=2) FreqItemset(items=['q', 'y', 'z'], freq=2) FreqItemset(items=['q', 't'], freq=2) FreqItemset(items=['q', 't', 'y'], freq=2) FreqItemset(items=['q', 't', 'y', 'x'], freq=2) FreqItemset(items=['q', 't', 'y', 'x', 'z'], freq=2) FreqItemset(items=['q', 't', 'y', 'z'], freq=2) FreqItemset(items=['q', 't', 'x'], freq=2) FreqItemset(items=['q', 't', 'x', 'z'], freq=2) FreqItemset(items=['q', 't', 'z'], freq=2) FreqItemset(items=['q', 'x'], freq=2) FreqItemset(items=['q', 'x', 'z'], freq=2) FreqItemset(items=['q', 'z'], freq=2) 频繁项集数量 54

接下来我们将频繁项集转换为字典，以便后续的置信度，支持度的计算。

freqDict = result.map(lambda x:[tuple(sorted(x[0])), x[1]]).collectAsMap() print(freqDict) {('q', 'y'): 2, ('s', 't'): 2, ('q', 'z'): 2, ('t', 'x', 'y', 'z'): 3, ('t', 'x'): 3, ('p', 'z'): 2, ('q',): 2, ('q', 't', 'z'): 2, ('x', 'z'): 3, ('s',): 3, ('y',): 3, ('r', 'x'): 2, ('s', 't', 'x'): 2, ('s', 't', 'x', 'y'): 2, ('s', 't', 'x', 'y', 'z'): 2, ('q', 't', 'x', 'z'): 2, ('q', 'x', 'z'): 2, ('s', 't', 'x', 'z'): 2, ('s', 't', 'z'): 2, ('x',): 4, ('t', 'x', 'y'): 3, ('s', 'z'): 2, ('q', 'y', 'z'): 2, ('t', 'z'): 3, ('s', 'x', 'y'): 2, ('q', 'x'): 2, ('q', 't', 'y', 'z'): 2, ('x', 'y', 'z'): 3, ('s', 'x'): 3, ('s', 't', 'y'): 2, ('r',): 3, ('q', 't', 'x'): 2, ('q', 't'): 2, ('x', 'y'): 3, ('q', 'x', 'y'): 2, ('y', 'z'): 3, ('t', 'y', 'z'): 3, ('q', 'x', 'y', 'z'): 2, ('p',): 2, ('s', 't', 'y', 'z'): 2, ('r', 'z'): 2, ('t', 'x', 'z'): 3, ('s', 'x', 'z'): 2, ('p', 'r', 'z'): 2, ('q', 't', 'y'): 2, ('s', 'y', 'z'): 2, ('q', 't', 'x', 'y', 'z'): 2, ('q', 't', 'x', 'y'): 2, ('s', 'x', 'y', 'z'): 2, ('t',): 3, ('t', 'y'): 3, ('s', 'y'): 2, ('z',): 5, ('p', 'r'): 2}

接下来的重点在于从频繁项集中挖掘出我们想要的规则。如果你使用的是pyspark.ml版本，置信度，支持度都不需要自己计算，而如果你使用的是pyspark.mlib版本，则需要自己写代码，计算规则的置信度与提升度。

以下介绍如何通过频繁项集挖掘规则：

对于每一个频繁项集，我们都需要计算出其非空子集，以及非空子集的补集。

以下代码可以计算一个集合的非空子集：

def subSet(listVariable): # 求列表所有非空真子集的函数 newList = [] for i in range(1, len(listVariable)): newList.extend(list(itertools.combinations(listVariable, i))) return newList print(subSet([1,2,3,4])) [(1,), (2,), (3,), (4,), (1, 2), (1, 3), (1, 4), (2, 3), (2, 4), (3, 4), (1, 2, 3), (1, 2, 4), (1, 3, 4), (2, 3, 4)] def computeConfidence(freqItemset): itemset = freqItemset[0] #频繁项集 freq = freqItemset[1] #该项集出现的频率 subItemset = subSet(itemset)#该频繁项集的所有非空子集 rules = [] for i in subItemset: complement = tuple(set(itemset).difference(set(i))) # 对于每一个非空子集取补集 itemLink = str(complement) + '->' + str(i) confidence = float(freq) / freqDict[tuple(sorted(i))] # 求置信度 lift = float(freq) * NUM_DATA / (freqDict[tuple(sorted(i))] * freqDict[tuple(sorted(complement))]) #求支持度 rule = [itemLink, freq, confidence,lift] rules.append(rule) return rules

对于计算得到的频繁项集，使用上述代码挖掘 出关联规则：

confidence = result.flatMap(computeConfidence) tmp_confidence = confidence.collect() for i in tmp_confidence: print(i) print('计算完成置信度',len(tmp_confidence)) ["('z',)", "('x',)", "('z',)->('x',)", 0.6, 0.9] ["('x',)", "('z',)", "('x',)->('z',)", 0.75, 0.9] ["('x',)", "('y',)", "('x',)->('y',)", 0.75, 1.5] ["('y',)", "('x',)", "('y',)->('x',)", 1.0, 1.5] ["('z', 'x')", "('y',)", "('z', 'x')->('y',)", 1.0, 2.0] ["('y', 'z')", "('x',)", "('y', 'z')->('x',)", 1.0, 1.5] ....... 省略部分结果 ["('t',)", "('q', 'z')", "('t',)->('q', 'z')", 0.6666666666666666, 2.0] ["('q',)", "('t', 'z')", "('q',)->('t', 'z')", 1.0, 2.0] ["('x',)", "('q',)", "('x',)->('q',)", 0.5, 1.5] ["('q',)", "('x',)", "('q',)->('x',)", 1.0, 1.5] ["('z', 'x')", "('q',)", "('z', 'x')->('q',)", 0.6666666666666666, 2.0] ["('z', 'q')", "('x',)", "('z', 'q')->('x',)", 1.0, 1.5] ["('q', 'x')", "('z',)", "('q', 'x')->('z',)", 1.0, 1.2] ["('z',)", "('q', 'x')", "('z',)->('q', 'x')", 0.4, 1.2] ["('x',)", "('q', 'z')", "('x',)->('q', 'z')", 0.5, 1.5] ["('q',)", "('x', 'z')", "('q',)->('x', 'z')", 1.0, 2.0] ["('z',)", "('q',)", "('z',)->('q',)", 0.4, 1.2] ["('q',)", "('z',)", "('q',)->('z',)", 1.0, 1.2] 计算完成置信度，所有规则数目 324 保留大于置信度大于0.5和支持度大于1的规则 minSupportConfidence = confidence.filter(lambda x: x[3] > 0.5).filter(lambda x: x[4]>1) #保留置信度大于0.5，支持度大于1的规则 for rules in (minSupportConfidence.collect()): print(rules)

至此，我们便挖掘出了所有符合条件的关联规则，需要的话我们可以将挖掘出的规则保存至mysql等数据库，以供这些规则的后续使用。

其中需要注意的是频繁模式挖掘的最小支持度的设定，如果设置的过低，会导致挖掘出大量的频繁项集，导致内存报错，后续的规则挖掘也变得异常缓慢。所以需要具体根据实际情况，设定最小支持度。

完整代码

最后附上使用pyspark.mllib和 pyspark.ml 两个版本的关联规则挖掘代码与数据，大家可以对比下跑一下程序。建议大家使用pyspark.ml 版本，因为这个版本的算法封装的更好，使用起来更加方便。 完整代码地址：https://github.com/taiguangxing/spark_associationrules