新手入門：Spark部署實戰(zhàn)入門

創(chuàng)新互聯(lián)公司是一家專業(yè)提供榆林企業(yè)網(wǎng)站建設(shè),專注與做網(wǎng)站、成都做網(wǎng)站、H5網(wǎng)站設(shè)計、小程序制作等業(yè)務。10年已為榆林眾多企業(yè)、政府機構(gòu)等服務。創(chuàng)新互聯(lián)專業(yè)網(wǎng)站建設(shè)公司優(yōu)惠進行中。

Spark簡介 

整體認識 

Apache Spark是一個圍繞速度、易用性和復雜分析構(gòu)建的大數(shù)據(jù)處理框架。最初在2009年由加州大學伯克利分校的AMPLab開發(fā)，并于2010年成為Apache的開源項目之一。 Spark在整個大數(shù)據(jù)系統(tǒng)中處于中間偏上層的地位，如下圖，對hadoop起到了補充作用： 

基本概念 

Fork/Join框架是Java7提供了的一個用于并行執(zhí)行任務的框架， 是一個把大任務分割成若干個小任務，最終匯總每個小任務結(jié)果后得到大任務結(jié)果的框架。

第一步分割任務。首先我們需要有一個fork類來把大任務分割成子任務，有可能子任務還是很大，所以還需要不停的分割，直到分割出的子任務足夠小。

第二步執(zhí)行任務并合并結(jié)果。分割的子任務分別放在雙端隊列里，然后幾個啟動線程分別從雙端隊列里獲取任務執(zhí)行。子任務執(zhí)行完的結(jié)果都統(tǒng)一放在一個隊列里，啟動一個線程從隊列里拿數(shù)據(jù)，然后合并這些數(shù)據(jù)。

具體可參考Fork/Join

核心概念

RDD(Resilient Distributed Dataset) 彈性分布數(shù)據(jù)集介紹 彈性分布式數(shù)據(jù)集（基于Matei的研究論文）或RDD是Spark框架中的核心概念。

可以將RDD視作數(shù)據(jù)庫中的一張表。其中可以保存任何類型的數(shù)據(jù)。Spark將數(shù)據(jù)存儲在不同分區(qū)上的RDD之中。 RDD可以幫助重新安排計算并優(yōu)化數(shù)據(jù)處理過程。

此外，它還具有容錯性，因為RDD知道如何重新創(chuàng)建和重新計算數(shù)據(jù)集。

RDD是不可變的。你可以用變換（Transformation）修改RDD，但是這個變換所返回的是一個全新的RDD，而原有的RDD仍然保持不變。

RDD支持兩種類型的操作： o 變換（Transformation） o 行動（Action） 變換：變換的返回值是一個新的RDD集合，而不是單個值。調(diào)用一個變換方法，不會有任何求值計算，它只獲取一個RDD作為參數(shù)，然后返回一個新的RDD。變換函數(shù)包括：map，filter，flatMap，groupByKey，reduceByKey，aggregateByKey，pipe和coalesce。

行動：行動操作計算并返回一個新的值。當在一個RDD對象上調(diào)用行動函數(shù)時，會在這一時刻計算全部的數(shù)據(jù)處理查詢并返回結(jié)果值。

行動操作包括：reduce，collect，count，first，take，countByKey以及foreach。 共享變量（Shared varialbes） o 廣播變量（Broadcast variables） o 累加器（Accumulators） Master/Worker/Driver/Executor

o Master：

1. 接受Worker的注冊請求，統(tǒng)籌記錄所有Worker的CPU、Memory等資源，并跟蹤Worker結(jié)點的活動狀態(tài)；2. 接受Driver中App的注冊請求(這個請求由Driver端的Client發(fā)出)，為App在Worker上分配CPU、Memory資源，生成后臺Executor進程；之后跟蹤Executor和App的活動狀態(tài)。 o Worker：負責接收Master的指示，為App創(chuàng)建Executor進程。Worker在Master和Executor之間起著橋梁作用，實際不會參與計算工作。 o Driver：負責用戶側(cè)邏輯處理。 o Executor：負責計算，接受并執(zhí)行由App劃分的Task任務，并將結(jié)果緩存在本地內(nèi)存或磁盤。

Spark部署

關(guān)于Spark的部署網(wǎng)上相關(guān)資料很多，這里進行歸納整理 部署環(huán)境 Ubuntu 14.04LTS Hadoop:2.7.0 Java JDK 1.8 Spark 1.6.1 Scala 2.11.8

Hadoop安裝

由于Spark會利用HDFS和YARN，所以需要提前配置Hadoop，配置教程可以參考： Setting up a Apache Hadoop 2.7 single node on Ubuntu 14.04 Hadoop安裝教程_單機/偽分布式配置_Hadoop2.6.0/Ubuntu14.04

Spark安裝

在安裝好Hadoop的基礎(chǔ)上，搭建Spark，配置教程參考：

Spark快速入門指南 – Spark安裝與基礎(chǔ)使用

scala安裝

Scala作為編寫Spark的源生語言，更新速度和支持情況肯定是最好的，而另一方面Scala本身語言中對于面向?qū)ο蠛秃瘮?shù)式編程兩種思想的糅合，使得該語言具有很多炫酷的語法糖，所以在使用Spark的過程中我采用了Scala語言進行開發(fā)。

Scala最終編譯成字節(jié)碼需要運行在JVM中，所以需要依托于jdk，需要部署jdk Eclipse作為一款開發(fā)Java的IDE神器，在Scala中當然也可以使用，有兩種方式: o Eclipse->Help->Install New Software安裝Scala Plugins o 下載官網(wǎng)已經(jīng)提供的集成好的Scala IDE 基于以上兩步已經(jīng)可以進行Scala開發(fā)，需要用到Scala自帶的SBT編譯的同學可以裝下Scala官網(wǎng)下載地址，本人一直使用Maven進行包管理就延續(xù)Maven的使用

簡單示例：WordCount（Spark Scala） 開發(fā)IDE：Eclipse Scala 包管理：Maven 開發(fā)語言：Scala

創(chuàng)建Maven項目

1. 跳過archetype項目模板的選擇

2. 下載模板pom.xml

3. 對maven項目添加Scala屬性： Right click on project -> configure - > Add Scala Nature.

4. 調(diào)整下Scala編譯器的版本，與Spark版本對應： Right click on project- > Go to properties -> Scala compiler -> update Scala installation version to 2.10.5

5. 從Build Path中移除Scala Library（由于在Maven中添加了Spark Core的依賴項，而Spark是依賴于Scala的，Scala的jar包已經(jīng)存在于Maven Dependency中）： Right click on the project -> Build path -> Configure build path and remove Scala Library Container.

6. 添加package包com.spark.sample

1. 創(chuàng)建Object WordCount和SimpleCount，用來作為Spark的兩個簡單示例 Spark Sample SimpleCount.scala

package com.spark.sample

import org.apache.spark.SparkConf import org.apache.spark.SparkContext

object SimpleCount { def main(args: Array[String]) { val conf = new SparkConf().setAppName("TrySparkStreaming").setMaster("local[2]") // Create spark context val sc = new SparkContext(conf) // val ssc = new StreamingContext(conf, Seconds(1)) // create streaming context

    val txtFile = "test"
    val txtData = sc.textFile(txtFile)
    txtData.cache()

    txtData.count()
    val wcData = txtData.flatMap { line => line.split(",") }.map { word => (word, 1) }.reduceByKey(_ + _)
    wcData.collect().foreach(println)

    sc.stop
}

}

WordCount.scala

package com.spark.sample import org.apache.spark.SparkConf import org.apache.spark.SparkContext import org.apache.spark.rdd.RDD.rddToPairRDDFunctions object WordCount { def main(args: Array[String]) = {

    //Start the Spark context
    val conf = new SparkConf()
        .setAppName("WordCount")
        .setMaster("local")
    val sc = new SparkContext(conf)

    //Read some example file to a test RDD
    val test = sc.textFile("input.txt")

    test.flatMap { line => //for each line
        line.split(" ") //split the line in word by word.
    }.map { word => //for each word
        (word, 1) //Return a key/value tuple, with the word as key and 1 as value
    }.reduceByKey(_ + _) //Sum all of the value with same key
        .saveAsTextFile("output.txt") //Save to a text file

    //Stop the Spark context
    sc.stop}

}

原理如下圖：

 參考文獻：

1. http://km.oa.com/group/2430/articles/show/181711?kmref=search&from_page=1&no=1&is_from_iso=1

2. http://spark.apache.org/docs/latest/programming-guide.html#resilient-distributed-datasets-rdds

3. http://www.infoq.com/cn/articles/apache-spark-introduction?utm_source=infoq_en&utm_medium=link_on_en_item&utm_campaign=item_in_other_langs

4. http://www.infoq.com/cn/articles/apache-spark-sql

5. http://www.infoq.com/cn/articles/apache-spark-streaming

6. http://www.devinline.com/2016/01/apache-spark-setup-in-eclipse-scala-ide.html

7. https://databricks.gitbooks.io/databricks-spark-reference-applications/content/

8. http://wuchong.me/blog/2015/04/06/spark-on-hbase-new-api/

9. http://colobu.com/2015/01/05/kafka-spark-streaming-integration-summary/

10. http://www.devinline.com/2016/01/apache-spark-setup-in-eclipse-scala-ide.html

作者：張景龍 暢移（上海）信息科技有限公司CTO，CCFYOCSEF上海委員，京東今夜酒店特價APP技術(shù)奠基人和首任CTO，中國第一代智能手機開發(fā)者。