Spark程式設計模型之莎士比亞文集詞頻統計

        前段時間因為學校的雲端計算比賽我無意間接觸到了Spark雲端計算框架，從此對其一發不可收拾，無論從其執行效率還有他的其他方面的架構都感覺到無比強大，作為一個雲端計算的解決方案他有著比hadoop更好的優越性。因為Spark我又接觸到程式語言屆的新貴Scala面向物件的函數語言程式設計語言，更是被它的巧妙所折服。

        那麼問題來了，請問什麼事Spark?       Spark是一個基於記憶體計算的開源的叢集計算系統,目的是讓資料分析更加快速。Spark非常小巧玲瓏,由加州伯克利大學AMP實驗室的Matei為主的小團隊所開發。使用的語言是Scala，專案的core部分的程式碼只有63個Scala檔案，非常短小精悍。

       那麼接下來討論Spark程式設計模型，Spark 應用程式有兩部分組成：
         – Driver
        – Executor
首先由Driver向叢集管理器（Cluster Manager）申請資源，Cluster Manager在給出資源之後,Spark 在worker節點啟動Executor,然後用Driver將jar包檔案傳給Executor,並將任務分割成一個個Task分配給Executor執行，Executor執行結束，提交匯總結束任務執行。

Spark進行程式設計主要是對它的RDD( Resilient Distributed Datasets，彈性分散式資料集）進行操作也就是

    其中RDD的操作主要變現在三方面：1.由Base到RDD，也就是我們可以將HDFS中的檔案或者本地的檔案轉換成RDD。

2.Transformation操作也就是RDD->RDD（RDD之間互相轉化的過程）

3.Action操作也就是RDD->driver or Base（RDD返還給Driver或者轉化成hfds（或者本地）的檔案過程）

一下是我總結的一些函式

因為這是我第一次寫關於Spark方面的部落格，那麼我們就拿Spark的一個比賽中非常簡單的題目作介紹。

莎士比亞文集詞頻統計並行化演算法
 環境描述： 本題目需要執行在 Apache Spark 1.0.1 環境下，使用 Java 或者 Scala 進行程式設計開發。
 題目描述： 在給定的莎士比亞文集上（多個檔案） ，根據規定的停詞表，統計 出現頻度最高的 100 個單詞。
 資料集： shakespear 文集， 具體下載地址見大賽網站 http://cloud.seu.edu.cn。
 停詞表： stopword.txt， 具體下載地址見大賽網站 http://cloud.seu.edu.cn。
 程式設計約束： 程式需要三個輸入引數，第一個為資料集路徑（即 shakespear 資料夾的路徑，資料夾中的檔名為固定檔名），第二個為停詞表路徑，第三 個為輸出檔案路徑。 輸出檔案的格式為：

每個單詞獨立一行。

首先說明一下題目的意思，這個題目就是統計一下資料集： shakespear 文集中除去停詞表中出現的單詞中出現頻率最高的100個。

我的過程如下：
  1.停詞儲存
    因為涉及的停詞不是很多，但是要注意去除停詞表後面的空格，把他們讀入記憶體處理空格後形成一個Scala停詞陣列.
  2.flatMap()
    首先我會把每一行出現的製表符，逗號，冒號，分號等一些特殊符號替換成空格，然後按照空格將字串進行分割.
  3.filter
    把出現在停詞陣列中的單詞和空字元除去
  4.map
    把上面經過過濾的單詞設為key，值設為1
  5.reduceByKey
    將所有相同的key進行增量累加
  6.獲得前一百個key
    從上面所得的map結果集中獲得key列表，然後從列表中獲取前100個keyall
  7.將結果儲存磁碟檔案

具體的Scala程式碼如下：

package com.zdx.spark

import org.apache.spark.{ SparkContext, SparkConf }
import org.apache.spark.SparkContext._
/**
 * Created by zdx on 14-11-8.
 */


  object ShaShiBiYa {
    def main(args: Array[String]) {
      if (args.length != 3) {
        System.err.println("Usage: ShaShiBiYa <file of poems> <file of stopWord> <file of output>")
        System.exit(1)
      }

      val conf = new SparkConf().setAppName("ShaShiBiYa")
      val sc = new SparkContext(conf)
      val rddpoems=sc.textFile(args(0))
      val rddstop=sc.textFile(args(1))
      val stop=rddstop.map(_.replaceAll(" +","")).collect
      val word2=rddpoems.flatMap(_.replaceAll("\\t|\\(|\\)|\\||\\.|\\,|\\:|\\[|\\]|\\?|\\--|\\;|\\!"," ").split(" +")).filter(stop.contains(_)==false).filter(_.equals("")==false)
      val result2=word2.map((_,1)).reduceByKey(_+_).map(x=>(x._2,x._1)).sortByKey(false).map(x=>(x._2,x._1))
      val keyall=result2.keys
      val key100=keyall.take(100)
      val rdd8=sc.parallelize(key100)
      rdd8.saveAsTextFile(args(2))
      sc.stop()
    }


}

在具體的Scala程式碼中我們可以看到，程式碼非常的簡潔，可見Scala是多麼的強大，哈哈。繼續學習，天天向上！！！