2017/2/20

Spark tutorial


要使用 Spark 之前,一般會先遇到 scala 這個語言的熟悉度的問題,當有了一定的語言程度後,再來就是 scala IDE 的選擇,目前的狀況,還是IDEA 會比 scala IDE for Eclipse 好用。接下來就是下載跟安裝 spark,然後進行 WordCount 的範例練習,以下記錄怎麼安裝與設定 stand alone 的 spark 開發環境。


要下載哪一個 spark 套件


當我們連結到 Download Apache Spark 時,首先遇到的問題,就是要下載哪一個 spark release 套件。


基本的原則如下:


如果要直接下載已經編譯好的 binary 套件,我們可以根據 Hadoop 的版本決定要下載哪一個,但如果像我們一樣,不打算安裝 Hadoop 就直接測試,就直接選最新版的 spark-1.6.1-bin-hadoop2.6.tgz 就好了,下載後解壓縮,馬上就可以使用 spark-shell,或直接取得 all-in-one 的 spark-assembly-1.6.1-hadoop2.6.0.jar 套件。


如果我們要編譯 source code,就下載預設的 1.6.1(Mar 09 2016) spark release,Package type 選擇 Source Code:spark-1.6.1.tgz


由於目前 spark 預設是使用 scala 2.10 版,使用預先編譯的 spark 就必須要使用 scala 2.10 版,如果像要改成 2.11,就一定要自己重新編譯 spark,目前 spark 的 JDBC component 還不支援 scala 2.11。


Building for Scala 2.11 有兩行指令說明如何將 spark 由 2.10 調整為 2.11,我們同時把 hadoop 版本改為 2.6。


./dev/change-scala-version.sh 2.11
mvn -Pyarn -Phadoop-2.6 -Dscala-2.11 -DskipTests clean package

編譯 spark 要花的時間很久,以我現在的環境花了 40 分鐘。


[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 39:33 min
[INFO] Finished at: 2016-04-29T09:23:03+08:00
[INFO] Final Memory: 452M/2703M

也可以使用 sbt 來編譯 spark,編譯後會得到 spark-assembly 的 jar。


sbt/sbt assembly

如果要修改 spark souce code,可以啟用增量編譯模式,避免每一次修改都要花很久的時間重新編譯。


export SPARK_PREPEND_CLASSES=true
sbt/sbt compile

unset SPARK_PREPEND_CLASSES

在 compile 前面加上 ~ 可以避免每一次都重開一次新的 sbt console


sbt/sbt ~ compile

可以用 sbt 或是 mvn 指令查閱 dependency map


sbt/sbt dependency-tree

mvn -DskipTests install
mvb dependency:tree

如果要設定 spark source 的開發環境,可以用以下的指令產生 IDEA project file


git clone https://github.com/apache/spark
sbt/sbt gen-idea

Spark 開發環境 in IDEA


  1. 在 IDEA 建立新的 scala project: sparktest

  2. 在 project 中建立一個 lib 目錄,把 spark-assembly-1.6.1-hadoop2.6.0.jar 放在那個目錄中

  3. 在 File -> Project Structure -> Libraries 點 "+",然後把 lib 目錄加入 project 中

  4. 取得一個文字檔的測試資料 pg5000.txt ,將檔案放在新建立的 data 目錄中

  5. 將 RunWordCount.scala 放在 src 目錄中,程式會計算 pg5000.txt 裡面每一個字出現的數量


    import org.apache.log4j.Logger
    import org.apache.log4j.Level
    import org.apache.spark.{ SparkConf, SparkContext }
    import org.apache.spark.rdd.RDD
    
    object RunWordCount {
      def main(args: Array[String]): Unit = {
    
        // 以這兩行設定不顯示 spark 內部的訊息
        Logger.getLogger("org").setLevel(Level.OFF)
        System.setProperty("spark.ui.showConsoleProgress", "false")
    
        // 清除 output folder
        FileUtils.deleteDirectory(new File("data/output"))
    
        println("執行RunWordCount")
    
        // 設定 application 提交到 MASTER 指向的 cluster 或是 local 執行的模式
        // local[4] 代表是在本地以 四核心的 CPU 執行
        val sc = new SparkContext(new SparkConf().setAppName("wordCount").setMaster("local[4]"))
    
        println("讀取文字檔...")
        val textFile = sc.textFile("data/pg5000.txt") 
    
        println("開始建立RDD...")
        // flapMap 是取出文字檔的每一行資料,並以 " " 進行 split,分成一個一個的 word
        // map 是將每一個 word 轉換成 (word, 1) 的 tuple
        // reduceByKey 會根據 word 這個 key,將後面的 1 加總起來,就會得到 (word, 數量) 的結果
        val countsRDD = textFile.flatMap(line => line.split(" "))
          .map(word => (word, 1))
          .reduceByKey(_ + _) 
    
        println("儲存結果至文字檔...")
        try {
          countsRDD.saveAsTextFile("data/output") 
          println("存檔成功")
        } catch {
          case e: Exception => println("輸出目錄已經存在,請先刪除原有目錄");
        }
    
      }
    }
  6. 我們可以直接在 IDEA 就執行這個測試程式


    執行RunWordCount
    Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
    16/04/29 16:28:50 INFO Slf4jLogger: Slf4jLogger started
    16/04/29 16:28:50 INFO Remoting: Starting remoting
    16/04/29 16:28:50 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriverActorSystem@192.168.1.151:56205]
    讀取文字檔...
    開始建立RDD...
    儲存結果至文字檔...
    存檔成功
    
    Process finished with exit code 0
  7. 最後產生的結果有三個檔案,其中 part-00000 及 part-00001 裡面存了每一個 word 的發生次數


    _SUCCESS
    part-00000
    part-00001

會產生兩個檔案的原因是因為,spark 本身是平行運算的工具,所以會自動產生多個 partitions。


如果需要將結果整合成一個檔案,就必須使用 coalesce,在程式的最後面,用 countsRDD.coalesce(1).saveAsTextFile 將結果輸出到新目錄,也會得到一個檔案的結果。


try {
      countsRDD.coalesce(1).saveAsTextFile("data/output2")
      println("存檔成功")
    } catch {
      case e: Exception => println("輸出目錄已經存在,請先刪除原有目錄");
    }

匯出程式


  1. 在 IDEA 選擇 "File" -> "Project Structure" -> "Artifact"

  2. 點擊 "+" -> "JAR" -> "From modules with dependencies"

  3. Main Class 填成 "RunWordCount",輸出目錄的最後面改為 "out"

  4. 選擇 "Build" -> "Build Artifacts",就能在 out 目錄取得 sparktest.jar 檔

  5. 這樣就能在另一台機器執行 sparktest


java -jar sparktest.jar

References


HADOOP+SPARK大數據巨量分析與機器學習整合開發實戰


Spark大資料分析實戰