基于Spark的关联规则挖掘算法并行化研究

许德心，李玲娟

(南京邮电大学 计算机学院，江苏 南京 210023)

0 引 言

随着信息技术的迅猛发展，数据量急剧增加。如何从海量的随机数据中挖掘出有价值的信息，已成为一个必须面对的课题，数据挖掘技术由此诞生。数据挖掘是从大量的、不完全的、有噪声、模糊的、随机的实际应用数据中，提取隐含在其中的、人们事先不知道但又潜在有用的信息和知识的过程[1-2]。而关联规则挖掘则是数据挖掘中一个非常重要的研究课题。关联规则挖掘的主要任务分为两项：从大量数据中找到频繁项集，根据频繁项集提取有价值的强关联规则。其中Apriori算法是其典型代表。该算法简单准确，但是面对急速增长的数据量，该算法对强关联规则的提取效率有待提高[3-5]。

Spark作为新一代的大数据运算框架，将数据载入内存，之后的迭代计算可以直接使用内存中的中间结果作运算，避免了从磁盘中频繁读取数据，拥有更高的执行效率，很适合运行迭代运算较多的数据挖掘与机器学习算法。基于此，文中研究了Apriori算法在Spark平台上的并行化方案，以提高算法提取强关联规则的准确性与实效性，并设计了实验对该方案的使用效果进行了检验。

1 Apriori算法原理

关联规则一般描述：项集→项集，如X→Y。

支持度(support)：表示X，Y同时出现的概率。关联规则X→Y的支持度可表示为：

(1)

置信度(confidence)：表示在X出现的情况下Y也出现的概率。

(2)

其中，δ(X)=|ti|X⊆ti,ti∈T|，是项集X出现的次数，ti表示某个事务的标识TID，T表示事务的集合。

频繁项集：是支持度大于最小支持度阈值的项集。

强关联规则：频繁项集中置信度大于最小置信度阈值的关联规则。

Apriori算法利用逐层搜索迭代的方法找出项集之间的关系，最终形成规则。这个过程由连接与剪枝组成，为了降低连接与剪枝的时间复杂度，需借助以下定理：

定理1：如果一个项集是频繁的，则其所有的子集也一定是频繁的。

定理2：如果一个项集是非频繁的，那么其所有的超集也一定是非频繁的。

Apriori算法步骤如下：

Step1：设定最小支持度s和最小置信度c。

Step2：根据数据集产生出候选1-项集，若其支持度大于等于最小支持度，那么它就是频繁1-项集。

Step3：频繁1-项集通过连接产生候选2-项集，再通过候选2-项集获得频繁2-项集。

Step4：不断迭代产生下一级候选项集，直到不再产生新的候选项集为止。

Step5：由频繁项集生成关联规则。具体过程是：根据频繁项集Fk，找出可以出现在规则右部大小为m的元素列表Hm，如果频繁项集可以移除大小为m的子集，且对于规则Fk-Hk→Hm的置信度大于最小置信度阈值，则该条规则为强关联规则[6-7]。在这个过程中，同样可以借助以下定理进行剪枝。

下面给出Apriori算法产生频繁项集及由频繁项集生成关联规则的伪代码，其中Ck为候选K-项集的集合，Fk为频繁K-项集的集合。

产生频繁项集：

1.K=1

2.Fk={i|i∈I∧δ{i}≥n*minsup|}

//找出频繁一项集

3.Repeat

4.K=K+1

5.ti

//由频繁项集生成下一级候选项集

6.for each transactiont∈Tdo

7.Ct=subset(Ck,t)

//确定所有属于t的候选项集

8.for each candidate itemsetsc∈Ct

9.c∈Ct

//项集c出现的次数+1

10.end for

11.end for

Fk={c|c∈Ck∧δ(c)≥N*minsup|}

//获取频繁K-项集

12.直到频繁K-项集为空

13.将所有的频繁项集取并集

生成关联规则：

1.k=|fk| //频繁项集的大小

2.m=|Hm| //可以出现在规则右部元素列表的大小

3.ifk>m+1 then //如果频繁项集可以移除大小为m的子集

4.Hm+1=apriori-gen(Hm)

//使用apriori-gen函数生成无重复组合

5.for eachhm+1∈Hm+1do //对每个元素进行处理

6.conf=δ(fk)/δ(fk-hm+1) //得出它们的置信度

7.if conf≥minconf then //如果大于最小置信度阈值

8.output the rule (fk-hm+1)→hm+1//输出关联规则

9.else deletehm+1fromHm+1

10.end if

11.end for

12.call ap-genrules(fk,Hm+1)

13.end if

定理3：如果规则X→Y不满足置信度阈值，那么对于X的子集X'，规则X'→Y也不满足置信度阈值。

从Apriori算法的执行过程可以看出该算法多次进行循环，产生了大量的候选项集，需要多次扫描数据库，这就需要很大的I/O负载。为此，借助Spark平台设计相应的并行化方案来解决这个问题。

2 Spark核心机制

Spark是由美国UC Berkeley的AMP实验室开发的基于内存计算的大数据并行计算框架，它是MapReduce的拓展，可用于构建大型的、低延迟的数据分析应用程序，性能优于Hadoop。Spark的设计遵循“一个软件栈满足不同应用场景”的理念，逐渐形成了一套完整的生态系统，具有一系列优势。

(1)运行速度快。Spark使用先进的DAG(directed acyclic graph，有向无环图)执行引擎，以支持循环数据流与内存计算，基于内存的执行速度可比Hadoop MapReduce快上百倍，基于磁盘的执行速度也能快十倍；

(2)容易使用。Spark支持Scala、Java、Python和R语言进行编程，简洁的API设计有助于用户轻松构建并行程序，并且可以通过Spark Shell进行交互式编程；

(3)通用性。Spark提供了完整而强大的技术栈，包括SQL查询、流式计算、机器学习和图算法组件，这些组件可以无缝整合在同一个应用中，足以应对复杂的计算；

(4)运行模式多样。Spark可运行于独立的集群模式中，或者运行于Hadoop中，也可运行于Amazon EC2等云环境中，并且可以访问HDFS、Cassandra、HBase、Hive等多种数据源。

Spark的核心是弹性分布式数据集(resilient distributed dataset，RDD)。一个RDD本质上是一个只读的分区记录集合，每个RDD可以分成多个分区，并且不同分区可以被保存到集群中不同的节点上，从而可以在各节点进行并行计算。RDD提供两种类型的操作，分别为行动(Action)和转换(Transformation)，前者用于计算并指定输出形式，后者指定RDD之间的依赖关系。Transformation操作都是延时执行的，它只是逻辑操作，不会进行具体计算，只有通过Action操作才会将RDD放入内存中求得结果值[8]。在RDD的设计中，数据只读，不可修改，如果需要修改数据，必须从父RDD转换到子RDD，由此在不同RDD之间建立了血缘关系。所以，RDD通过RDD父子血缘关系重新计算得到丢失的分区来实现容错，无需回滚整个系统，避免了数据复制的高开销。此外，数据在内存中的多个RDD操作之间进行传递，避免了读写磁盘的开销。

Spark的运行架构如图1所示。

Spark运行架构包含集群资源管理器、运行作业任务的工作节点(Worker)、每个应用的任务控制节点(Driver)和每个工作节点负责具体任务的执行进程(Executor)。当一个任务被用户提交时，Driver节点会创建一个SparkContext，由SparkContext向资源管理器(Cluster Manager)申请资源；资源分配完毕后，Spark会启动Worker上负责执行具体任务的进程Executor，并会将任务分发给Executor；计算完成后，Worker会将结果发回Driver，然后释放相关资源。Spark的Executor利用多线程来执行具体任务，减少任务的启动开销；其中有一个BlockManager存储模块，会将内存和磁盘共同作为存储设备，当进行多轮迭代计算时(Apriori算法是典型的例子)，可以将中间结果存到这个存储模块里，下次需要时，就可以直接读该存储模块里的数据，而不需要对HDFS等文件系统读写，从而大大减少了I/O的开销[9-10]。

3 Apriori算法基于Spark的并行化方案设计

由以上分析可知，Spark平台提供的机制与Apriori算法多次迭代计算的需求十分契合。

文中基于Spark的特性和“分而治之”的思想，通过把事务数据集分发给多个子节点和RDD转换来设计Apriori算法基于Spark的并行化方案，用局部查找频繁项集、剪枝代替全局操作，避免全局查找产生的内存负担。

Apriori算法的并行化步骤如下：

(1)Spark的配置与数据源的读取。

首先将原始的交易数据存放在分布式文件系统HDFS上，并将其转化为压缩矩阵。此时Spark驱动程序会读取相关的配置文件生成SparkConf对象，接着创建SparkContext对象用来连接访问Spark集群，进一步采用textFile算子扫描HDFS上的压缩矩阵，根据压缩矩阵数据创建RDD。如前文大数据编程模型所述，Spark并行框架计算流程实际上是通过数据集转化为待处理RDD，然后根据待处理RDD进行一系列的Transformation操作得到新的RDD，最后调用Action操作求得结果值。由于RDD是Spark大数据框架的最大特点，也是其高运行效率的重要原因，因此在进行计算任务时，RDD的数量要与Spark集群为程序分配的计算资源相匹配，否则会导致Spark框架计算效率降低，程序并发度下降。

(2)Apriori算法的并行化。

基于Spark平台的Apriori算法分为两部分：频繁项集生成和关联规则提取[11-13]。其中频繁项集的生成使用MapReduce思想来实现，通过Map操作计算候选项集的局部支持度计数，Reduce计算候选项集的全局支持度计数，具体流程如图2所示。

图2 Apriori算法基于Spark的并行化流程

Apriori算法并行化实现的关键是迭代调用Transformation和Action操作，每次迭代中利用上一次的迭代结果进行求解[14]。为了实现并行化，每个Worker节点通过多线程方式对其RDD分区中的数据使用Apriori算法计算。首先从分布式文件系统中获取数据并创建弹性分布式数据集RDD_1，接着在单机的情况下统计项的个数来计算频繁1-项集RDD_2，判断候选2-项集是否存在，若存在，则对数据集使用map和cache操作，将数据集缓存到内存中，同时进入频繁K-项集循环；在循环中，根据Worker节点上的频繁1-项集使用reduceByKey算子过滤数据集，生成局部候选K-项集RDD_3，并求出局部候选K-项集的支持度计数；RDD_4进行连接生成全局支持度计数，产生频繁K-项集，再判断候选K+1项集是否存在，重复迭代，直到候选K+1项集不存在，则频繁K-项集生成完毕。根据频繁项集提取关联规则，其中置信度大于最小置信度阈值的关联规则为强关联规则[15]。

4 实验与结果分析

(1)实验环境与数据。

为了验证基于Spark的并行化Apriori算法的执行效率，设计了相关实验。实验环境包含四个节点，1个Master节点，3个Worker节点。CPU版本为Intel Core i7-4710HQ，每个CPU都拥有4个752.3 MB/s的处理单元。Master节点与Worker节点内存均为4 G。Spark版本为1.6.1；Spark运行的操作系统为CentOS6.5；Java版本为JDK1.8.0_144；Scala版本为2.11.0。

实验采用数据生成器生成的数据集，大小为4 MB，模拟了网上商城共10 W次交易记录，将该数据集分为2 W、4 W、6 W、8 W、10 W进行多次实验，以确保实验结果的准确性。这些数据集中的强关联规则数量分别是：3 150，5 718，9 463，12 620，15 780。

使用的软件为IDEA+Spark，用Scala语言进行算法实现。

(2)实验结果与分析。

对数据进行采样，将支持度置为0.8，做了两组实验，统计运行时间。

实验一：对比在单机Apriori算法和集群上并行化Apriori算法的执行速度，以及两者对强关联规则挖掘的准确性。结果分别如图3和表1所示。

由实验结果可以看出，Spark集群上Apriori算法实施挖掘所需的时间明显少于单机所需的时间；同时，集群上挖掘出的强关联规则与单机上挖掘出的强关联规则保持一致，说明该并行化方案在保证准确性的前提下，具有良好的时效性。

图3 单机和Spark集群上的时间效率对比

数据量单机强关联规则数集群强关联规则数2 W3 1503 1504 W5 7185 7186 W9 4639 4638 W12 62012 62010 W15 78015 780

实验二：测试并行化Apriori算法运行时间随节点增加的变化情况。用了10 W条数据集，分别测量节点数为1、2、3、4时的Apriori算法产生强关联规则的时间，结果如图4所示。

图4 不同节点数下并行化Apriori算法的执行时间

由图4可知，随着节点个数增多，算法执行时间不断缩短。另外，得出的强关联规则数是15 780，因此节点的个数并不影响算法的准确率。

5 结束语

以提高关联规则挖掘的时效性为目标，设计了一种Apriori算法基于Spark集群的并行化方案，并用数据生成器生成的数据集对算法的执行效率进行了验证。实验结果表明，基于Spark的并行化Apriori算法在面对大量数据的情况下，具有时间效率上的明显优势，并且随着执行节点的增加，并行化效果更好。