Spark分布式机器学习源码分析：频繁模式挖掘

Posted 2021-04-05 雨云飞

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Spark分布式机器学习源码分析：频繁模式挖掘相关的知识，希望对你有一定的参考价值。

Spark是一个极为优秀的大数据框架，在大数据批处理上基本无人能敌，流处理上也有一席之地，机器学习则是当前正火热AI人工智能的驱动引擎，在大数据场景下如何发挥AI技术成为优秀的大数据挖掘工程师必备技能。本文结合机器学习思想与Spark框架代码结构来实现分布式机器学习过程，希望与大家一起学习进步～

本文采用的组件版本为：Ubuntu 19.10、Jdk 1.8.0_241、Scala 2.11.12、Hadoop 3.2.1、Spark 2.4.5，老规矩先开启一系列Hadoop、Spark服务与Spark-shell窗口：

挖掘频繁项（FP），项集，子序列或其他子结构通常是分析大规模数据集的第一步，而这是多年来数据挖掘中的活跃研究主题。spark.mllib提供了FP-growth的并行实现，这是一种用于挖掘频繁项集的流行算法。

FP增长算法

FP增长算法在Han等人的论文中进行了描述，该算法在不生成候选者的情况下挖掘频繁模式，其中“ FP”代表频繁模式。给定交易数据集，FP增长的第一步是计算项目频率并识别频繁项目。与为相同目的设计的类似Apriori的算法不同，FP-growth的第二步使用后缀树（FP-tree）结构对交易进行编码，而无需显式生成候选集，这通常成本较高。第二步之后，可以从FP树中提取频繁项集。在spark.mllib中，我们实现了称为PFP的FP-growth的并行版本，如Li et al。，PFP：并行FP-growth用于查询推荐中所述。PFP根据事务的后缀分配增长的FP树的工作，因此比单机实现更具可伸缩性。我们请用户参考这些文件以获取更多详细信息。

spark.mllib的FP-growth实现采用以下（超）参数：

minSupport：对某个项目集的最低支持，该项目集被标识为频繁。例如，如果某项出现在5个事务中的3个，则其支持率为3/5 = 0.6。
numPartitions：用于分发工作的分区数。

FP示例

FPGrowth实现FP-growth算法。它采用事务的RDD，其中每个事务都是通用类型的项目托盘。如何从事务中挖掘超出的项目集和关联规则。

import org.apache.spark.mllib.fpm.FPGrowthimport org.apache.spark.rdd.RDDval data = sc.textFile("data/mllib/sample_fpgrowth.txt")val transactions: RDD[Array[String]] = data.map(s => s.trim.split(' '))val fpg = new FPGrowth().setMinSupport(0.2).setNumPartitions(10)val model = fpg.run(transactions)model.freqItemsets.collect().foreach { itemset => println(s"${itemset.items.mkString("[", ",", "]")},${itemset.freq}")}val minConfidence = 0.8model.generateAssociationRules(minConfidence).collect().foreach { rule => println(s"${rule.antecedent.mkString("[", ",", "]")}=> " + s"${rule.consequent .mkString("[", ",", "]")},${rule.confidence}")}

关联规则

数据挖掘是⼀项从⼤量的记录数据中提取有价值的、⼈们感兴趣的知识，这些知识是隐含的、事先未知的有⽤信息，提取的知识⼀般可表⽰为概念(Concepts)、规则(Rules)、规律(Regular ides)、模式(Patterns)等形式。

关联规则是当前数据挖掘研究的主要⽅法之⼀，它反映⼀个事物与其他事物之间的相互依存性和关联性。如果两个或者多个事物之间存在⼀定的关联关系，那么，其中⼀个事物就能够通过其他事物预测到。

典型的关联规则发现问题是对超市中的货篮数据（Market Basket）进行分析。通过发现顾客放⼊货篮中的不同商品之间的关系来分析顾客的购买习惯。

AssociationRules实现了并行规则生成算法，用于构造具有单个项目的规则。

import org.apache.spark.mllib.fpm.AssociationRulesimport org.apache.spark.mllib.fpm.FPGrowth.FreqItemsetval freqItemsets = sc.parallelize(Seq( new FreqItemset(Array("a"), 15L), new FreqItemset(Array("b"), 35L), new FreqItemset(Array("a", "b"), 12L)))val ar = new AssociationRules().setMinConfidence(0.8)val results = ar.run(freqItemsets)results.collect().foreach { rule =>println(s"[${rule.antecedent.mkString(",")}=>${rule.consequent.mkString(",")} ]" +    s" ${rule.confidence}")}

序列模式挖掘

PrefixSpan（序列模式挖掘）是在Pei等人的《通过模式增长来挖掘顺序模式：PrefixSpan方法》中描述的顺序模式挖掘算法。spark.mllib的PrefixSpan实现采用以下参数：

minSupport：被视为频繁顺序模式所需的最低支持。
maxPatternLength：频繁顺序模式的最大长度。任何超出此长度的频繁模式都不会包含在结果中。
maxLocalProjDBSize：在开始对投影数据库进行本地迭代处理之前，前缀投影数据库中允许的最大项目数。该参数应根据执行程序的大小进行调整。

以下示例说明了在序列上运行的PrefixSpan（使用与Pei等人相同的表示法）：

import org.apache.spark.mllib.fpm.PrefixSpanval sequences = sc.parallelize(Seq( Array(Array(1, 2), Array(3)), Array(Array(1), Array(3, 2), Array(1, 2)), Array(Array(1, 2), Array(5)), Array(Array(6))), 2).cache()val prefixSpan = new PrefixSpan().setMinSupport(0.5).setMaxPatternLength(5)val model = prefixSpan.run(sequences)model.freqSequences.collect().foreach { freqSequence => println( s"${freqSequence.sequence.map(_.mkString("[", ", ", "]")).mkString("[", ", ", "]")}," + s" ${freqSequence.freq}")}