程序员人生 网站导航

K-频繁项集挖掘并行化算法(Apriori算法在Spark上的实现)

栏目:服务器时间:2015-01-10 08:59:30

大家好,下面为大家分享的实战案例是K-频繁相机发掘并行化算法。相信从事数据发掘相干工作的同学对频繁项集的相干算法

比较了解,这里我们用Apriori算法及其优化算法实现。

首先说1下实验结果。对2G,1800W条记录的数据,我们用了18秒就算完了1⑻频繁项集的发掘。应当还算不错。

给出题目:

本题的较第4题难度更大。我们在写程序的时候1定要注意写出的程序是并行化的,而不是只在client上运行的单机程序。否

则你的算法效力将让你跌破眼镜。另外还需要对算法做相干优化。在这里主要和大家交换1下算法思路和相干优化。

对Apriori算法的实现在这里不做过量赘述,百度1下大片大片。在Spark上实现这个算法的时候主要分为两个阶段第1阶段

是1个整体的循环求出每一个项集的阶段,第2阶段主要是针对第i个项集求出第i+1项集的候选集的阶段。

对这个算法可以做以下优化:

  1. 视察!这点很重要,经过视察可以发现有大量重复的数据,所谓方向不对努力白费也是这个道理,首先需要紧缩重复的数据。不然会做许多无用功。
  2. 设计算法的时候1定要注意是并行化的,大家可能很疑惑,Spark不就是并行化的么?可是你1不谨慎可能就写成只在client端运行的算法了。
  3. 由于数据量比较大,切记多使用数据持久化和BroadCast广播变量对中间数据进行相应处理。
  4. 数据结构的优化,BitSet是1种优秀的数据结构他只需1位就能够存储以个整形数,对所给出的数据都是整数的情况特别适用。 
下面给出算法实现源码:
  • import scala.util.control.Breaks._
  • import scala.collection.mutable.ArrayBuffer
  • import java.util.BitSet
  • import org.apache.spark.SparkContext
  • import org.apache.spark.SparkContext._
  • import org.apache.spark._


  • object FrequentItemset {
  •   def main(args: Array[String]) {
  •     if (args.length != 2) {
  •       println("USage:<Datapath> <Output>")
  •     }
  •     //initial SparkContext
  •     val sc = new SparkContext()
  •     val SUPPORT_NUM = 15278611 //Transactions total is num=17974836, SUPPORT_NUM = num*0.85
  •     val TRANSACITON_NUM = 17974836.0
  •     val K = 8


  •     //All transactions after removing transaction ID, and here we combine the same transactions.
  •     val transactions = sc.textFile(args(0)).map(line =>
  •       line.substring(line.indexOf(" ") + 1).trim).map((_, 1)).reduceByKey(_ + _).map(line => {
  •       val bitSet = new BitSet()
  •       val ss = line._1.split(" ")
  •       for (i <- 0 until ss.length) {
  •         bitSet.set(ss(i).toInt, true)
  •       }
  •       (bitSet, line._2)
  •     }).cache()


  •     //To get 1 frequent itemset, here, fi represents frequent itemset
  •     var fi = transactions.flatMap { line =>
  •       val tmp = new ArrayBuffer[(String, Int)]
  •       for (i <- 0 until line._1.size()) {
  •         if (line._1.get(i)) tmp += ((i.toString, line._2))
  •       }
  •       tmp
  •     }.reduceByKey(_ + _).filter(line1 => line1._2 >= SUPPORT_NUM).cache()
  •     val result = fi.map(line => line._1 + ":" + line._2 / TRANSACITON_NUM)
  •     result.saveAsTextFile(args(1) + "/result⑴")


  •     for (i <- 2 to K) {
  •       val candiateFI = getCandiateFI(fi.map(_._1).collect(), i)
  •       val bccFI = sc.broadcast(candiateFI)
  •       //To get the final frequent itemset
  •       fi = transactions.flatMap { line =>
  •         val tmp = new ArrayBuffer[(String, Int)]()
  •         //To check if each itemset of candiateFI in transactions
  •         bccFI.value.foreach { itemset =>
  •           val itemArray = itemset.split(",")
  •           var count = 0
  •           for (item <- itemArray) if (line._1.get(item.toInt)) count += 1
  •           if (count == itemArray.size) tmp += ((itemset, line._2))
  •         }
  •         tmp
  •       }.reduceByKey(_ + _).filter(_._2 >= SUPPORT_NUM).cache()
  •       val result = fi.map(line => line._1 + ":" + line._2 / TRANSACITON_NUM)
  •       result.saveAsTextFile(args(1) + "/result-" + i)
  •       bccFI.unpersist()
  •     }
  •   }


  •   //To get the candiate k frequent itemset from k⑴ frequent itemset
  •   def getCandiateFI(f: Array[String], tag: Int) = {
  •     val separator = ","
  •     val arrayBuffer = ArrayBuffer[String]()
  •     for(i <- 0 until f.length;j <- i + 1 until f.length){
  •       var tmp = ""
  •       if(2 == tag) tmp = (f(i) + "," + f(j)).split(",").sortWith((a,b) => a.toInt <= b.toInt).reduce(_+","+_)
  •       else {
  •         if (f(i).substring(0, f(i).lastIndexOf(',')).equals(f(j).substring(0, f(j).lastIndexOf(',')))) {
  •           tmp = (f(i) + f(j).substring(f(j).lastIndexOf(','))).split(",").sortWith((a, b) => a.toInt <= b.toInt).reduce(_ + "," + _)
  •         }
  •       }
  •       var hasInfrequentSubItem = false //To filter the item which has infrequent subitem
  •       if (!tmp.equals("")) {
  •         val arrayTmp = tmp.split(separator)
  •         breakable {
  •           for (i <- 0 until arrayTmp.size) {
  •             var subItem = ""
  •             for (j <- 0 until arrayTmp.size) {
  •               if (j != i) subItem += arrayTmp(j) + separator
  •             }
  •             //To remove the separator "," in the end of the item
  •             subItem = subItem.substring(0, subItem.lastIndexOf(separator))
  •             if (!f.contains(subItem)) {
  •               hasInfrequentSubItem = true
  •               break
  •             }
  •           }
  •         } //breakable
  •       }
  •       else hasInfrequentSubItem = true
  •       //If itemset has no sub inftequent itemset, then put it into candiateFI
  •       if (!hasInfrequentSubItem) arrayBuffer += (tmp)
  •     } //for
  •     arrayBuffer.toArray
  •   }
  • }
先写到这里,欢迎大家提出相干的建议或意见。
            (by老杨,转载请注明出处)


------分隔线----------------------------
------分隔线----------------------------

最新技术推荐