K-頻繁項集挖掘并行化算法（Apriori算法在Spark上的實現）

大家好，下面為大家分享的實戰(zhàn)案例是K-頻繁相機發(fā)掘并行化算法。相信從事數據發(fā)掘相干工作的同學對頻繁項集的相干算法

首先說1下實驗結果。對2G，1800W條記錄的數據，我們用了18秒就算完了1⑻頻繁項集的發(fā)掘。應當還算不錯。

給出題目：

本題的較第4題難度更大。我們在寫程序的時候1定要注意寫出的程序是并行化的，而不是只在client上運行的單機程序。否

則你的算法效力將讓你跌破眼鏡。另外還需要對算法做相干優(yōu)化。在這里主要和大家交換1下算法思路和相干優(yōu)化。

對Apriori算法的實現在這里不做過量贅述，百度1下大片大片。在Spark上實現這個算法的時候主要分為兩個階段第1階段

是1個整體的循環(huán)求出每一個項集的階段，第2階段主要是針對第i個項集求出第i+1項集的候選集的階段。

對這個算法可以做以下優(yōu)化：

1. 視察！這點很重要，經過視察可以發(fā)現有大量重復的數據，所謂方向不對努力白費也是這個道理，首先需要緊縮重復的數據。不然會做許多無用功。
2. 設計算法的時候1定要注意是并行化的，大家可能很疑惑，Spark不就是并行化的么？可是你1不謹慎可能就寫成只在client端運行的算法了。
3. 由于數據量比較大，切記多使用數據持久化和BroadCast廣播變量對中間數據進行相應處理。
4. 數據結構的優(yōu)化，BitSet是1種優(yōu)秀的數據結構他只需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老楊，轉載請注明出處）

生活不易，碼農辛苦
如果您覺得本網站對您的學習有所幫助,可以手機掃描二維碼進行捐贈