2. 程式人生 > >基於spark實現並行化Apriori演算法

基於spark實現並行化Apriori演算法

阿新 • • 發佈：2018-12-21

詳細程式碼我已上傳到github：click me

一、 實驗要求

        在 Spark2.3 平臺上實現 Apriori 頻繁項集挖掘的並行化演算法。要求程式利用 Spark 進
行平行計算。

二、演算法設計

2.1 設計思路

  • 變數定義
    • D為資料集，設Lk是k項頻繁項集，Ck是k項候選集，每一行資料定義為一筆交易（transaction），交易中的每個商品為項item。
    • 支援度: support, 即該項集在資料集D中出現的次數
  • 演算法流程
    • 單機Apriori演算法的主要步驟如下：
      1. 獲取輸入資料，產生頻繁1項集，以及和I作為候選集，掃描資料集D，獲取候選集C1的支援度，並找出最小支援度min_sup的元素作為頻繁1項集L1.
      2. 掃描資料集D，獲取候選集Ck的支援度，並找出其中滿足最小支援度的元素作為頻繁k項集Lk
      3. 通過頻繁k項集Lk產生k+1候選集Ck+1
      4. 通過迭代步驟2和3,直到找不到k+1項集結束

並行化設計的思路主要是考慮將對於支援度計數的過程使用wordcount來進行統計。

2.2 並行化演算法設計

        Apriori演算法產生頻繁項集有兩個特點：第一，它是逐層的，即從頻繁1-項集到頻繁k-項集；第二，它使用產生-測試的策略來發現頻繁項，每次迭代後都由前一次產生的頻繁項來產生新的候選項，然後對新產生的候選項集進行支援度計數得到新的頻繁項集。根據演算法的特點，我們將演算法分為兩個階段：
如下圖1.1演算法的並行化框架圖，主節點每次迭代時需要將候選項集以廣播的形式分發到每個從節點，每個從節點收到之後進行一些列的操作得到新的頻繁項集，如此反覆直至求得最大頻繁項集。

圖1.1 並行化框架圖

  • 階段1：從HDFS上獲取原始的資料集SparkRDD，載入到分散式記憶體中。掃描所有的RDD事務，進行支援度計數，產生頻繁1-項集；如圖1.2所示為Ap演算法並行化第一階段的Lineage圖。

    圖1.2 Apriori演算法並行化第一階段的Lineage圖

    原始事務集由flatMap函式去讀取事務，並將所有的事務轉化為Spark RDD並cache到分散式記憶體中。接下來，在每一個事務中執行flatMap函式來獲取所有的Items項集，之後執行map函式，發射<Item, 1>的key/value形式，接下來執行reduceByKey函式統計每一個候選1-項集的支援度，最後並利用事先設好的支援度閾值進行剪枝，所有超過支援度閾值的項集將會生成頻繁1-項集，下面給出了第一階段的演算法虛擬碼

  • 階段2：在這個階段，不斷迭代使用頻繁k-項集去產生頻繁k+1項集

    圖1.3 Apriori演算法並行化第二階段的Lineage圖

    如圖1.3所示，首先讀取頻繁k-項集並且以<itemSet, Count>的形式將其儲存為Spark RDD。接下來，從頻繁k-項集得到候選(k+1)-項集。為加速從候選項集中查詢(k+1)-項集的過程，將候選(k+1)-項集存放在雜湊表中，並將其broadcast到每個worker節點。接下來，通過flatMap函式獲取每個候選項集在原始事務集中的支援度，進一步對每個候選項使用map函式得到<ItemSet, 1>，之後通過reduceBykey函式蒐集每個事務的最終的支援度計數，最後利用實現設定好的支援度閾值剪枝，支援度大於等於最小閾值的頻繁項集將會以key/value的形式輸出，作為頻繁(k+1)-項集，下面給出了演算法第二階段的虛擬碼。

2.3 程式設計與效能分析

  • 讀取資料集，按空格劃分每行內容，並用HashSet儲存，方便後期求子集以及一些集合操作

    // 將輸入資料分割槽，由於後面要頻繁使用。因此快取起來
val transations = sc.textFile(input, num)
        .map(x => {
            val content = x.split("\\s+")
            val tmp = new HashSet[String]
            for (i <- 0 until content.length) {
                tmp.add(content(i))
            }
            tmp
        }).cache()

  • 根據支援度和資料總量計算頻繁項閾值，便於後期統計集合頻度後直接對比

    // 計算頻繁項閾值
val numRecords = transations.count().toDouble
val freqThre = numRecords * support
println("frequency threshold：" + freqThre)

  • 計算頻繁1項集用於後續的迴圈迭代計算

    // 計算頻繁1項集
val oneFreqSet = transations
        .flatMap(_.seq)
        .map(x => (x, 1))
        .reduceByKey(_ + _)
        .filter(x => x._2 >= freqThre)
        .map(x => (HashSet(x._1), x._2 / numRecords))

  • 利用上一輪迭代計算生成的頻繁k項集來構造候選k+1項集，然後通過比頻繁項閾值比對篩選出頻繁k+1項集。這裡有一點要注意的，由於從檔案讀入的源資料transaction被劃分在各個partition上，而候選集candidates要與transaction中每條記錄比對來統計頻度，因此需要spark呼叫broadcast將候選集廣播到每個partition上

    // 生成頻繁項的候選集
val candidates = generateCandidates(preFreSets, round)
// 將候選項集廣播到各個分割槽
val broadcastCandidates = sc.broadcast(candidates)

//複雜度：len(transactions) * len(candidates) * round * transaction項的平均長度
//這裡的len(transaction)是指各個partition上transaction的平均長度
val curFreqSet = transations
    .flatMap(x => verifyCandidates(x, broadcastCandidates.value))
    .reduceByKey(_ + _)
    .filter(x => x._2 >= freqThre)

// 寫入頻繁round項集結果到hdfs
curFreqSet.map(a => {
    val out = a._1.mkString(",") + ":" + (a._2 / numRecords).toString
    out
}).saveAsTextFile(output + "/" + infileName + "freqset-" + round)

// 生成頻繁round-Itemsets，用於下一輪迭代生成候選集
preFreSets = curFreqSet.collect().map(x => x._1)

    第round輪迭代，由候選項集生成頻繁項集的複雜度：len(transactions) * len(candidates) * round * transaction項的平均長度，這裡的len(transaction)是指各個partition上transaction的平均長度，儘管我們通過提高併發度的方式將複雜度的稍微將了一些，可是演算法的整體複雜度還是很高，特別是當源資料集很大時，這樣查表式地驗證候選集很費時，有考慮將項集索引，但是如果全部項集都存那這個儲存開銷太大了，目前沒有很好的優化思路，時間有限也沒有進一步深入怎麼優化這一步了。

  • 對於候選集生成方法generateCandidates的具體實現，我們首先拆分上一輪頻繁項集preFreSets中的每個項再合併成一個元素集，相當於一個詞彙表，然後遍歷preFreSets中每個項，如果該項中不包含元素表中的某個元素，則將該元素與該項合併成一個候選項。具體實現如下：

    def generateCandidates(preFreSets : Array[HashSet[String]], curRound: Int): Array[HashSet[String]] = {
    // 複雜度：len(elements) * len(preFrestats)^2 * curRound^2
    val elements = preFreSets.reduce((a,b) => a.union(b))
    val canSets = preFreSets.flatMap( t => for (ele <- elements if(!t.contains(ele))) yield t.+(ele) ).distinct
    canSets.filter( set => {
        val iter = set.subsets(curRound - 1)
        var flag = true
        while (iter.hasNext && flag){
            flag = preFreSets.contains(iter.next())
        }
        flag
    })
}

    但是這個過程複雜度太高：len(elements) * len(preFrestats)^2 * curRound^2，當資料來源中元素過多，迭代更深以後，這個複雜度將變得讓人難以接受，花了大量的時間再前一輪的候選項集中驗證候選項，需要想一個辦法來避免順序式的查表，但限於時間有限，這個地方沒有深入展開研究怎麼優化。

2.4 關聯規則的實現

  • 利用當前這一輪迭代生成的頻繁項集curFreqSet來計算關聯規則，利用curFreqSet建立頻繁項索引freqSetIndex，同統計候選項頻度的原因一樣，我們需要將freqSetIndex廣播到各個partition以統計規則A->B左項A的頻度，再利用freqSetIndex索引AB頻繁項的頻度即可計算規則A->B的置信度，然後與設定的置信度對比即可篩選出需要的關聯規則，程式碼實現如下：

    // 生成關聯規則
val asst1 = System.nanoTime()
// 建立頻繁round-Itemsets的索引Map
val freqSetIndex = HashMap[HashSet[String], Int]()
curFreqSet.collect().foreach(fs => freqSetIndex.put(fs._1, fs._2))
// 將頻繁round-Itemsets的索引Map廣播到各個partition
val broadcastCurFreqSet = sc.broadcast(freqSetIndex)
// 生成所有可能的關聯規則，然後篩選出置信度>=confidence的關聯規則
val associationRules = transations
        .flatMap(x => verifyRules(x, broadcastCurFreqSet.value.keys.toArray, round))
        .reduceByKey(_ + _)
        .map(x => ((x._1._1, x._1._2), broadcastCurFreqSet.value.get(x._1._1.union(x._1._2)).getOrElse(0) * 1.0 / x._2))
        .filter(x => x._2 >= confidence)

  • 對於規則構造verifyRules的具體實現，我們通過遍歷規則左項的長度來構造，具體實現如下：

    def verifyRules(transaction: HashSet[String], candidates: Array[HashSet[String]], curRound: Int): Array[((HashSet[String], HashSet[String]), Int)] = {
    // yield會根據第一個迴圈型別返回對應的型別，這裡的candidates是Array，因此返回的也是Array型別
    for {
        set <- candidates

        i <- 1 until curRound

        iter = set.subsets(i)
        l <- iter
        if (l.subsetOf(transaction))
        r = set.diff(l)
    } yield ((l, r), 1)
}

三、實驗環境、執行方式及結果

3.1 環境

spark分散式環境的安裝

  1. 在本地配置好java,scala,hadoop(Spark會用到hadoop的hdfs)
    版本： jdk 1.8.0_161, scala 2.11.8, hadoop 2.7.5

  2. 在spark官網下載spark-2.3.0-bin-hadoop2.7,解壓安裝

 tar -zxvf spark-2.3.0-bin-hadoop2.7 -C ~/bigdata/spark
  1. 配置環境變數,並使環境變數生效
$ vim ~/.bashrc
# Spark Environment Variables
export JAVA_HOME=~/bigdata/java/jdk1.8.0_161
export JRE_HOME=${JAVA_HOME}/jre
export SCALA_HOME=~/bigdata/scala/scala-2.11.8
export HADOOP_HOME=~/bigdata/hadoop/hadoop-2.7.5
export SPARK_HOME=~/bigdata/spark/spark-2.3.0-bin-hadoop2.7

export CLASS_PATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib

export PATH=${JAVA_HOME}/bin:${SCALA_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${SPARK_HOME}/bin:${SPARK_HOME}/sbin:$PATH

$ source /etc/profile
  1. 配置，spark的配置檔案位於$SPARK_HOME/conf目錄下，需要修改的檔案有spark-env.sh, spark-defaluts.conf和slaves。
$ cd ~/bigdata/spark/spark-2.3.0-bin-hadoop2.7/conf
$ cp spark-env.sh.template spark-env.sh
$ vim spark-env,sh
# spark-env.sh configuration
export JAVA_HOME=~/bigdata/java/jdk1.8.0_161
export SCALA_HOME=~/bigdata/scala/scala-2.11.8
export SPARK_HOME=~/bigdata/spark/spark-2.3.0-bin-hadoop2.7
export HADOOP_HOME=~/bigdata/hadoop/hadoop-2.7.5
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
export PATH=$PATH:$JAVA_HOME/bin:$SCALA_HOME/bin:$SPARK_HOME/bin:$HADOOP_HOME/bin
export SPARK_MASTER_IP=slave103

$ cp spark-defaluts.conf.template spark-defaults.conf
$ vim saprk-defaults.conf
# spark-defaults.conf configuration
spark.executor.extraJavaOptions    -XX:+PrintGCDetails -XX:+PrintGCTimeStamps
spark.eventLog.enabled             true
spark.eventLog.dir             hdfs://slave103:9000/spark_event
spark.yarn.historyServer.address   slave103:18080
spark.history.fs.logDirectory      hdfs://slave103:9000/history_log
spark.serializer                   org.apache.spark.serializer.KryoSerializer
spark.executor.memory              2g
spark.executor.cores               2
spark.driver.memory                2g
spark.driver.cores                 2
spark.yarn.am.memory               2g
spark.yarn.am.cores                2
spark.local.dir                    /tmp/sparklocaldir
spark.yarn.jars                    hdfs://slave103:9000/spark/jars/*                 

$ cp slaves.template slaves
$ vim slaves
# slaves configuration(主機名在/etc/hosts中配置)
slave101
slave103
  1. 啟動和停止
# 啟動
$ bash $SPARK_HOME/sbin/start-all.sh
# 停止
$ bash $SPARK_HOME/sbin/stop-all.sh
  1. 啟動hadoop和spark執行jps命令，顯示的程序如下圖3.1和3.2所示：
圖3.1 主節點jvm程序
圖3.2 從節點jvm程序

3.2 jar包執行方式

假設輸入資料檔案為chess.dat，shell下執行方式如下：

spark-submit --class main.scala.Apriori.distributed.Apriori --master spark://slave103:7077 --conf spark.driver.memory=4g --conf spark.executor.cores=2 original-MapReduce-1.0.jar input/apriori/chess.dat output 0.8 20 24 0.9
# jar後面的引數說明：輸入檔案 輸出目錄 支援度 迭代輪數 併發度即partition數目 置信度

3.3 結果

  • 測試connect.dat資料集生成頻繁項集的執行時間，圖3.3是單機版的，圖3.4是並行版的：
圖3.3 單機版執行時間
圖3.4 並行版執行時間

  • 在chess.dat資料集上測試並行版本的頻繁項集生成和關聯規則挖掘的執行時間如下：

  • chess.dat頻繁項集生成結果

  • chess.dat關聯規則挖掘結果

四、WebUI執行報告

相關推薦

基於spark實現並行化Apriori演算法

詳細程式碼我已上傳到github：click me 一、 實驗要求         在 Spark2.3 平臺上實現 Apriori 頻繁項集挖掘的並行化演算法。要求程式利用 Spark 進 行平行計算。 二、演算法設計

基於Spark實現推薦演算法-1:推薦演算法簡介

個性化推薦系統簡介 個性化推薦系統的定義在 1997 年由 Resnick 和 Varian 提出:利用網際網路向用戶提供信 息和建議，幫助使用者選擇產品，或模擬售貨員幫助使用者完成購買行為的系統 。通常推薦 由三個要素組成:推薦演算法、使用者、候選推薦專案。

基於Spark實現推薦演算法-4:基於物品的協同過濾(實現篇)

演算法設計與實現 基於物品的協同過濾又稱Item-Based CF. 基於Spark的Item-Based CF演算法其實現原理和步驟與經典方法基本一致，不同的地方主要在於具體步驟內的並行化計算。 相似度演算法 在Spark MLlib中提供了餘弦相

OpenMp實現並行化

隱式 必須 board chunk 並行化 ads 任務 方法 sso 前言 昨天，往arm上移植opencv程序，發現運行速度很慢。觀察資源監視器發現只有一個核處於高負荷（總共4核），遂考慮到需要多核計算。OpenMp和MPI是常用並行計算庫，OpenMP相對

[原始碼和文件分享]基於C++實現的LZW壓縮演算法

1 特點 基於C++實現的LZW壓縮演算法，特點如下所示： 使用stl::map鍵值對作為字典儲存 感覺算是簡單的檔案操作 字典無限長，字典自生長。但是字典只能解析儲存ascii編碼之類存在，中文符號之類的碰到就掛 2 邏輯設計 2.1 總體思路

Future和Callable，CompletionService實現並行化

Callable和Future實現並行化 Future表示一個任務的生命週期，並可以判斷是否完成和取消。 介面 方法如下： boolean cancel(boolean);取消任務 V get();獲取結果，阻塞等待 V get(long, TimeUnit)

基於spark實現表的join操作

1. 自連線 假設存在如下檔案： [root@bluejoe0 ~]# cat categories.csv 1,生活用品,0 2,數碼用品,1 3,手機,2 4,華為Mate7,3 每一行的格式為：類別ID，類別名稱，父類ID 現在欲輸出每個類別

一行 Python 實現並行化 -- 日常多執行緒操作的新思路

春節坐在回家的火車上百無聊賴，偶然看到 Parallelism in one line 這篇在 Hacker News 和 reddit 上都評論過百的文章，順手譯出，enjoy:-) http://www.zhangzhibo.net/2014/02/01/

基於spark實現kmeans的更新聚類中心的關鍵程式碼

/** * 這個方法作用是返回被給樣本的所屬聚類中心的索引，及其到這個聚類中心的距離， * 其中class VectorWithNorm(val vector: Vector, val norm: Double)這是VectorWithNorm

Aprior並行化演算法在Spark上的實現

本文為大家分享的Spark實戰案例是K-頻繁項集挖掘——Apriori並行化演算法的實現。關聯資料探勘、頻繁項集挖掘的常用演算法有Apriori,Fp-growth以及eclat演算法。這裡我使用Apriori演算法進行頻繁項集挖掘。Apriori演算法於200

車間排程_基於DFS和貪心的可隨機化求解演算法_C++實現

一、作業車間排程問題描述     作業車間排程問題（Job Shop Scheduling, JSP）是最經典的幾個NP-hard問題之一。其應用領域極其廣泛，涉及航母排程，機場飛機排程，港口碼頭貨船排程，汽車加工流水線等。     JS

基於Spark MLlib 在IDEA環境實現Kmeans 演算法

import org.apache.spark.{SparkContext, SparkConf} import org.apache.spark.mllib.clustering.{KMeans, KMeansModel} import org.apache.spark.mllib.linalg.Ve

基於Spark的移動使用者主要活動地點的挖掘演算法實現以及JavaEE技術整合

本演算法基於Spark計算引擎，能夠從海量的手機基站資料中挖據出使用者的主要活動地點，比如工作地點和居住地點。實現好挖掘演算法之後，通過JavaEE來整合上面的演算法，讓使用者能夠通過簡單的Web UI就能夠操作使用該演算法，同時為使用者提供了視覺化資料的功能。 1、移動使

基於spark的svm演算法實現

支援向量機SVM(Support Vector Machine)是一種有監督的學習模型，它的核心有兩個：一、核函式(kernel trick)；二、序列最小優化演算法SMO（Sequential minimal optimization）是John Platt在1996年釋

pyspark實現Apriori演算法、迴圈迭代、並行處理

from pyspark import SparkContext myDat=[ [ 1, 3, 4,5 ], [ 2, 3, 5 ], [ 1, 2, 3,4, 5 ], [ 2,3,4, 5 ] ] sc = SparkContext( 'local', 'pyspa

基於Java FX實現的視覺化排序演算法

Java FX是jdk7出來的新特性，它和swing一個最大的不同之處就是Java FX可以使用fxml來配合Scene Builder來構建介面，這個非常類似於Android使用xml佈局檔案來設計介面。當然，如果你不想使用fxml，你可以直接使用程式碼來動態生成介面，這個

基於DNS實現智能化訪問網站

智能dns 智能DNS比傳統的DNS具有更高的智能選擇化，可以根據不同的ip地址來判斷出此ip來自哪個區域，從而達到根據ip可以訪問到本區域的服務器，這樣的好處就是減少網絡延遲以及提升訪問速度。不過，現在一般中小型企業在沒有太大能力的情況下都會把DNS托付給第三方給予管理（CDN: Content Del

GIS資訊關聯規則挖掘——Apriori演算法的實現（下）

上篇說明了原理，這篇就直接上核心程式碼了~ 程式碼比較長，所以理解可能有點麻煩，核心思路就是計算選擇的維度後遍歷資料，逐步進行迴圈計算置信度，並淘汰每次迴圈後的最低值。 這裡有一點要注意的，我一開始想用arraylist構造一個堆疊結構進行資料遍歷的儲存跟計算，因為這樣效率比較高。

GIS資訊關聯規則挖掘——Apriori演算法的實現（上）

最近閒著無聊沒啥課，幫讀master的朋友做了一個桌面端的GIS系統，主要功能是景區管理。 其中有個核心功能挺有意思的，就是統計所有景區受損設施的所有致損型別和每個型別具體包含的致損因子後，計算致損因子之間的關聯規則，然後可以根據使用者選定的致損型別組合計算出其景區設施造成損害的概率。（有點

資料探勘：基於Spark+HanLP實現影視評論關鍵詞抽取(1)

1. 背景 近日專案要求基於爬取的影視評論資訊，抽取影視的關鍵字資訊。考慮到影視評論資料量較大，因此採用Spark處理框架。關鍵詞提取的處理主要包含分詞+演算法抽取兩部分。目前分詞工具包較為主流的，包括哈工大的LTP以及HanLP，而關鍵詞的抽取演算法較多，包括TF-IDF、TextRank、互資訊等。本次