環(huán)境配置，Spark實現(xiàn)WordCount

本人準(zhǔn)備參加騰訊實習(xí)，有關(guān)大數(shù)據(jù)與機器學(xué)習(xí)。由于本人對大數(shù)據(jù)一無所知，因此準(zhǔn)備由Spark作為切入口開始自學(xué)，一步步完成機器學(xué)習(xí)各個算法在Spark上的應(yīng)用。自學(xué)過程中的點點滴滴都會記錄在簡書上，希望可以與大家交流，共同學(xué)習(xí)。

配環(huán)境永遠(yuǎn)是開始學(xué)習(xí)一個新領(lǐng)域最難的一部分，我花了兩天時間配置成功了MacOS下的Spark開發(fā)環(huán)境，實現(xiàn)了基于Scala與sbt的WordCount，下面來一步步把步驟記錄下來。

第1步：配置sbt在IntelliJ下編程環(huán)境

打開terminal

查看java版本，由于MacOS自帶java，因此無需安裝

$ java -version

安裝sbt，這是編譯scala的工具

$ brew install sbt

查看sbt與scala信息

$ sbt about

下載安裝IntelliJ

安裝Scala Plugin：打開IntelliJ，在選擇項目界面，選擇Configure → Plugins → Install JetBrains Plugins，搜索Scala并安裝

選擇默認(rèn)SDK：Configure → Project defaults → Project structure，SDK選擇Java1.8

至此scala在IntelliJ下的開發(fā)環(huán)境配置完畢

第2步：配置Spark工具包

下載Spark：下載地址，注意如果已經(jīng)安裝了Hadoop的話要下載對應(yīng)的版本，下面的命令可以查看Hadoop版本

$ hadoop version

下載完畢后解壓并將其放在一個目錄下，假設(shè)放在/usr/shar/spark-2.1.0-bin-hadoop2.7，那么我們往環(huán)境變量中添加Spark方便以后使用

$ vim .bash_profile

加入一行，保存后重啟terminal即可

export SPARK_HOME=/usr/shar/spark-2.1.0-bin-hadoop2.7

至此，Spark環(huán)境配置完畢，是不是非常方便

第3步：命令行形式操控Spark

(1) Python Spark

terminal中執(zhí)行命令

$ $SPARK_HOME/bin/pyspark

看到帥氣的Spark logo就表示已經(jīng)成功了

美中不足的是自帶的python shell沒有自動補全等功能，使用ipython可以完美解決

首先，安裝ipython

$ pip install ipython

運行Spark

$ PYSPARK_DRIVER_PYTHON=ipython $SPARK_HOME/bin/pyspark

讓我們來使用一些Spark的API來嘗試一些命令

>>> lines = sc.textFile("README.md") # 創(chuàng)建一個名為lines的RDD
>>> lines.count() # 統(tǒng)計RDD中的元素個數(shù) 
127
>>> lines.first()

(2) Scala Spark Shell

$ $SPARK_HOME/bin/spark-shell

同樣完成一下行數(shù)統(tǒng)計的小應(yīng)用

scala> val lines = sc.textFile("README.md") // 創(chuàng)建一個名為lines的RDD
lines: spark.RDD[String] = MappedRDD[...]
scala> lines.count() // 統(tǒng)計RDD中的元素個數(shù) 
res0: Long = 127
scala> lines.first() // 這個RDD中的第一個元素，也就是README.md的第一行 
res1: String = # Apache Spark

第4步：構(gòu)建Spark獨立應(yīng)用，WordCount

上面的是shell形式下調(diào)用Spark，而現(xiàn)在進(jìn)入更為重要的建立獨立項目，我看了很多教程，但是每個教程都有一步兩步講的含糊不清，或者就是版本太老，留下了許多坑?，F(xiàn)在我總結(jié)了一個可以跑通的例子。

首先，IntelliJ下創(chuàng)建sbt項目：打開IntelliJ → Create New Project → Scala → sbt → ProjectName = wordcount → Create

修改build.sbt，在最后加入一行Spark的包。注意scalaVersion一定要改成2.11，因為Spark2.1.0是基于Scala2.11的，默認(rèn)的2.12會報錯！

name := "wordcount"
version := "1.0"
scalaVersion := "2.11.7"
libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

讓我們先來看一下sbt項目的目錄結(jié)構(gòu)

├── build.sbt
├── project
│   ├── build.properties
│   ├── plugins.sbt
│   ├── project
│   └── target
├── src
│   ├── main
│   │   ├── java
│   │   ├── resources
│   │   ├── scala
│   │   └── scala-2.12
│   └── test
│       ├── java
│       ├── resources
│       ├── scala
│       └── scala-2.12
└── target
    ├── resolution-cache
    ├── scala-2.12
    └── streams

我們需要寫的代碼主要放在/src/main/scala里面

下一步，我們開始寫我們的代碼，具體細(xì)節(jié)不用深究，本章節(jié)只是為了配通環(huán)境

新建目錄/src/main/scala/com/oreilly/learningsparkexamples/mini/scala

添加第一個文件/src/main/scala/com/oreilly/learningsparkexamples/mini/scala/BasicMap.scala

/**
 * Illustrates a simple map in Scala
 */
package com.oreilly.learningsparkexamples.scala

import org.apache.spark._

object BasicMap {
    def main(args: Array[String]) {
      val master = args.length match {
        case x: Int if x > 0 => args(0)
        case _ => "local"
      }
      val sc = new SparkContext(master, "BasicMap", System.getenv("SPARK_HOME"))
      val input = sc.parallelize(List(1,2,3,4))
      val result = input.map(x => x*x)
      println(result.collect().mkString(","))
    }
}

添加第二個文件/src/main/scala/com/oreilly/learningsparkexamples/mini/scala/WordCount.scala

/**
 * Illustrates flatMap + countByValue for wordcount.
 */
package com.oreilly.learningsparkexamples.mini.scala

import org.apache.spark._
import org.apache.spark.SparkContext._

object WordCount {
    def main(args: Array[String]) {
      val inputFile = args(0)
      val outputFile = args(1)
      val conf = new SparkConf().setAppName("wordCount")
      // Create a Scala Spark Context.
      val sc = new SparkContext(conf)
      // Load our input data.
      val input =  sc.textFile(inputFile)
      // Split up into words.
      val words = input.flatMap(line => line.split(" "))
      // Transform into word and count.
      val counts = words.map(word => (word, 1)).reduceByKey{case (x, y) => x + y}
      // Save the word count back out to a text file, causing evaluation.
      counts.saveAsTextFile(outputFile)
    }
}

點擊右上角的Build Project圖標(biāo)就編譯成功了，如果沒有報錯，那么恭喜你，環(huán)境配置成功了。

第5步：使用spark-submit來運行應(yīng)用

spark-submit腳本可以為我們配置 Spark 所要用到的一系列環(huán)境變量。

首先需要將我們編譯好的項目打包，最方便的方式就是進(jìn)入wordcount目錄下，輸入

$ sbt package

打包好的文件就在/wordcount/target/scala-2.11/wordcount_2.11-1.0.jar

接下來就是利用Spark為我們提供的spark-submit來運行應(yīng)用了，進(jìn)入wordcount目錄下

$ $SPARK_HOME/bin/spark-submit \
--class com.oreilly.learningsparkexamples.mini.scala.WordCount  \
./target/scala-2.11/wc_2.11-1.0.jar \
./input.txt ./wordcounts

下面來簡單解釋一下上面的命令，--class為使用的Class，后面為jar包的路徑，最后兩個為wordcount的兩個參數(shù)，分別為輸入文件，和輸出文件路徑

我們的輸入文件\wordcount\input.txt是這樣的

one two three four
four five six
one five six
one one three

運行后，如果成功會在\wordcount\wordcounts\part-00000中看到

(two,1)
(one,4)
(six,2)
(three,2)
(five,2)
(four,2)

至此，我們的整個環(huán)境都配置成功啦，有問題請留言

參考資料

Spark官方文檔Quick-start

用SBT編譯Spark的WordCount程序

Big Data Analysis with Scala and Spark 洛桑聯(lián)邦理工學(xué)院 - Coursera