spark介紹

Spark是個通用的叢集計算框架，通過將大量資料集計算任務分配到多臺計算機上，提供高效記憶體計算。如果你熟悉Hadoop，那麼你知道分散式計算框架要解決兩個問題：如何分發資料和如何分發計算。Hadoop使用HDFS來解決分散式資料問題，MapReduce計算正規化提供有效的分散式計算。類似的，Spark擁有多種語言的函數語言程式設計API，提供了除map和reduce之外更多的運算子，這些操作是通過一個稱作彈性分散式資料集(resilient distributed datasets, RDDs)的分散式資料框架進行的。

本質上，RDD是種程式設計抽象，代表可以跨機器進行分割的只讀物件集合。RDD可以從一個繼承結構（lineage）重建（因此可以容錯），通過並行操作訪問，可以讀寫HDFS或S3這樣的分散式儲存，更重要的是，可以快取到worker節點的記憶體中進行立即重用。由於RDD可以被快取在記憶體中，Spark對迭代應用特別有效，因為這些應用中，資料是在整個演算法運算過程中都可以被重用。大多數機器學習和最優化演算法都是迭代的，使得Spark對資料科學來說是個非常有效的工具。另外，由於Spark非常快，可以通過類似Python REPL的命令列提示符互動式訪問。

Spark庫本身包含很多應用元素，這些元素可以用到大部分大資料應用中，其中包括對大資料進行類似SQL查詢的支援，機器學習和圖演算法，甚至對實時流資料的支援。

核心元件如下：

· Spark Core：包含Spark的基本功能；尤其是定義RDD的API、操作以及這兩者上的動作。其他Spark的庫都是構建在RDD和Spark Core之上的。

· Spark SQL：提供通過Apache Hive的SQL變體Hive查詢語言（HiveQL）與Spark進行互動的API。每個資料庫表被當做一個RDD，Spark SQL查詢被轉換為Spark操作。對熟悉Hive和HiveQL的人，Spark可以拿來就用。

· Spark Streaming：允許對實時資料流進行處理和控制。很多實時資料庫（如Apache Store）可以處理實時資料。Spark Streaming允許程式能夠像普通RDD一樣處理實時資料。

· MLlib：一個常用機器學習演算法庫，演算法被實現為對RDD的Spark操作。這個庫包含可擴充套件的學習演算法，比如分類、迴歸等需要對大量資料集進行迭代的操作。之前可選的大資料機器學習庫Mahout，將會轉到Spark，並在未來實現。

· GraphX：控制圖、並行圖操作和計算的一組演算法和工具的集合。GraphX擴充套件了RDD API，包含控制圖、建立子圖、訪問路徑上所有頂點的操作。

spark2.0釋出後，官網中將spark相關的第三方包放到libraries中 ：其中包含spark-als，mllib-grid-search等一些機器學習方法。

Spark mllib

MLlib 是spark的可以擴充套件的機器學習庫，由以下部分組成：通用的學習演算法和工具類，包括分類，迴歸，聚類，協同過濾，降維，當然也包括調優的部分。

· Data types  

· Basic statistics  (基本統計) 

o summary statistics 概括統計

o correlations 相關性

o stratified sampling 分層取樣

o hypothesis testing 假設檢驗

o random data generation 隨機數生成

· Classification and regression   (分類一般針對離散型資料而言的，迴歸是針對連續型資料的。本質上是一樣的) 

o linear models (SVMs, logistic regression, linear regression) 線性模型（支援向量機，邏輯迴歸，線性迴歸）

o naive Bayes 貝葉斯演算法

o decision trees  決策樹

o ensembles of trees (Random Forests and Gradient-Boosted Trees) 多種樹（隨機森林和梯度增強樹）

· Collaborative filtering   協同過濾 

o alternating least squares (ALS)  (交替最小二乘法(ALS) )

· Clustering  聚類 

o k-means k均值演算法

· Dimensionality reduction   (降維) 

o singular value decomposition (SVD)  奇異值分解

o principal component analysis (PCA) 主成分分析

· Feature extraction and transformation 特徵提取和轉化

· Optimization (developer) 優化部分 

o stochastic gradient descent 隨機梯度下降

o limited-memory BFGS (L-BFGS) 短時記憶的BFGS (擬牛頓法中的一種,解決非線性問題)

Spark提供了四種語言的程式設計API，如下：

Finally, full API documentation is available in Scala, Java, Python and R.

RDD概念:

RDD，全稱為Resilient Distributed Datasets(彈性分散式集合)，是一個容錯的、並行的資料結構，可以讓使用者顯式地將資料儲存到磁碟和記憶體中，並能控制資料的分割槽。同時，RDD還提供了一組豐富的操作（運算元）來操作這些資料。運算元包括Transformation（filter，map）和action，在這些操作中，諸如map、flatMap、filter等轉換操作實現了monad模式，很好地契合了Scala的集合操作。除此之外，RDD還提供了諸如join、groupBy、reduceByKey等更為方便的操作（注意，reduceByKey是action，而非 transformation），以支援常見的資料運算。

RDD可以看做是一個數組，資料的元素分佈在叢集中的各個機器上。Spark操作的物件是RDD，運算元操作會對RDD進行轉換，當RDD遇到aciton運算元時，將之前的所有運算元形成一個有向無環圖（DAG）RDD Graph，再在Spark中轉化為Job，提交到叢集執行。            

RDD儲存以partition為單位，和hdfs的block概念相同。（與hive 的partition概念不同，hive的是邏輯分割槽，比如按照日期分割槽）                                                                                                       

參考：《Spark大資料處理-技術、應用於效能優化》

安裝

1， 安裝scala

解壓tar -zxvf scala-2.11.7.tgz

配置scala環境變數,vi /etc/profile

#set the scala environment

export SCALA_HOME=/home/hadoop/Spark/scala-2.11.7

PATH=${SCALA_HOME}/bin:$PATH

使檔案生效：source /etc/profile

2， 安裝spark

Spark的安裝和配置比較簡單，

這裡直接下載帶有hadoop依賴的版本spark-*-bin-hadoop2.6.tgz，(spark-*-bin-without-hadoop是預編譯的無hadoop的版本，需要手動將hadoop依賴jar包進行拷貝)：

解壓tar -zxvf spark-1.5.0-bin-hadoop2.6.tgz

配置conf/spark-env.sh檔案：

cp spark-env.sh.template spark-env.sh

vi conf/spark-env.sh

export SCALA_HOME=/home/hadoop/Spark/scala-2.11.7//necessary

export SPARK_WORKER_MEMORY=1g

export SPARK_MASTER_IP=192.168.40.128

export MASTER=spark://192.168.40.128:7077

引數SPARK_WORKER_MEMORY決定在每一個Worker節點上可用的最大記憶體，正價這個數值可以在記憶體中快取更多的資料，但是一定要給Slave的作業系統和其他服務預留足夠的記憶體，這裡的單位可以使m,g,數值不能為小數，否則報錯。配置MASTER，否則會造成Slave無法註冊主機錯誤。

配置slaves檔案:

cp slaves.template slaves

vi conf/slaves

ubuntu2

spark的master需要免密碼登入到slaves上，所以保證安裝了ssh：

[ubuntu1]:ssh ubuntu2

Welcome to Ubuntu 14.04.3 LTS (GNU/Linux 3.19.0-26-generic x86_64)

l  Documentation:  https://help.ubuntu.com/

System information disabled due to load higher than 1.0

Last login: Fri Sep 25 18:07:58 2015 from 192.168.40.1

[email protected]:~$ jps

6377 Jps

[email protected]:~$ ssh ubuntu1

[email protected]'s password:

在所有worker上安裝並配置Spark

既然master上的這個檔案件已經配置好了，把它拷貝到所有的worker。注意，三臺機器spark所在目錄必須一致，因為master會登陸到worker上執行命令，master認為worker的spark路徑與自己一樣。

$ cd

$ scp -r spark-0.7.2 [email protected]:~

3，啟動叢集

啟動spark

./sbin/start-all.sh

關閉spark

./sbin/stop-all.sh

檢測是否安裝成功:

遇到問題，在 slave節點上，提示JAVA_HOME is not set，但是使用命令檢視:

echo $JAVA_HOME是存在的。

這裡解決方法是將export JAVA_HOME語句寫到spark-env.sh檔案中。

檢視日誌在logs下的.out檔案中。

啟動完hadoop後：

通過webUI檢視叢集狀態：

Interactive Analysis with the Spark Shell

首先將README.rm檔案上傳到hadoop的/user/hadoop/目錄下：

hadoop fs -put README.md /user/hadoop

Spark-shell help

執行./spark-shell --help　檢視spark-shell命令幫助：

 --master MASTER_URL         spark://host:port, mesos://host:port, yarn, or local.

  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or

                              on one of the worker machines inside the cluster ("cluster")

                              (Default: client).

  --class CLASS_NAME          Your application's main class (for Java / Scala apps).

  --name NAME                 A name of your application.

  --jars JARS                 Comma-separated list of local jars to include on the driver

                              and executor classpaths.

  --packages                  Comma-separated list of maven coordinates of jars to include

                              on the driver and executor classpaths. Will search the local

                              maven repo, then maven central and any additional remote

                              repositories given by --repositories. The format for the

                              coordinates should be groupId:artifactId:version.

  --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while

                              resolving the dependencies provided in --packages to avoid

                              dependency conflicts.

  --repositories              Comma-separated list of additional remote repositories to

                              search for the maven coordinates given with --packages.

  --py-files PY_FILES         Comma-separated list of .zip, .egg, or .py files to place

                              on the PYTHONPATH for Python apps.

  --files FILES               Comma-separated list of files to be placed in the working

                              directory of each executor.

  --conf PROP=VALUE           Arbitrary Spark configuration property.

  --properties-file FILE      Path to a file from which to load extra properties. If not

                              specified, this will look for conf/spark-defaults.conf.

  --driver-memory MEM         Memory for driver (e.g. 1000M, 2G) (Default: 1024M).

  --driver-java-options       Extra Java options to pass to the driver.

  --driver-library-path       Extra library path entries to pass to the driver.

  --driver-class-path         Extra class path entries to pass to the driver. Note that

                              jars added with --jars are automatically included in the

                              classpath.

  --executor-memory MEM       Memory per executor (e.g. 1000M, 2G) (Default: 1G).

  --proxy-user NAME           User to impersonate when submitting the application.

  --help, -h                  Show this help message and exit

  --verbose, -v               Print additional debug output

  --version,                  Print the version of current Spark

 Spark standalone with cluster deploy mode only:

  --driver-cores NUM          Cores for driver (Default: 1).

 Spark standalone or Mesos with cluster deploy mode only:

  --supervise                 If given, restarts the driver on failure.

  --kill SUBMISSION_ID        If given, kills the driver specified.

  --status SUBMISSION_ID      If given, requests the status of the driver specified.

 Spark standalone and Mesos only:

  --total-executor-cores NUM  Total cores for all executors.

 Spark standalone and YARN only:

  --executor-cores NUM        Number of cores per executor. (Default: 1 in YARN mode,

                              or all available cores on the worker in standalone mode)

 YARN-only:

  --driver-cores NUM          Number of cores used by the driver, only in cluster mode

                              (Default: 1).

  --queue QUEUE_NAME          The YARN queue to submit to (Default: "default").

  --num-executors NUM         Number of executors to launch (Default: 2).

  --archives ARCHIVES         Comma separated list of archives to be extracted into the

                              working directory of each executor.

  --principal PRINCIPAL       Principal to be used to login to KDC, while running on

                              secure HDFS.

  --keytab KEYTAB             The full path to the file that contains the keytab for the

                              principal specified above. This keytab will be copied to

                              the node running the Application Master via the Secure

                              Distributed Cache, for renewing the login tickets and the

                              delegation tokens periodically.

Python spark shell

這裡執行python 的spark shell：

./bin/pyspark

執行：

textFile=sc.textFile("README.md")//預設是從/user/hadoop目錄下讀取的

>>> textFile.count()

15/09/28 16:18:42 INFO mapred.FileInputFormat: Total input paths to process : 1

15/09/28 16:18:49 INFO spark.SparkContext: Starting job: count at <stdin>:1

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Got job 0 (count at <stdin>:1) with 2 output partitions

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Final stage: ResultStage 0(count at <stdin>:1)

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Parents of final stage: List()

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Missing parents: List()

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Submitting ResultStage 0 (PythonRDD[2] at count at <stdin>:1), which has no missing parents

15/09/28 16:18:49 INFO storage.MemoryStore: ensureFreeSpace(6064) called with curMem=145215, maxMem=560497950

15/09/28 16:18:49 INFO storage.MemoryStore: Block broadcast_1 stored as values in memory (estimated size 5.9 KB, free 534.4 MB)

15/09/28 16:18:49 INFO storage.MemoryStore: ensureFreeSpace(3559) called with curMem=151279, maxMem=560497950

15/09/28 16:18:49 INFO storage.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 3.5 KB, free 534.4 MB)

15/09/28 16:18:49 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.40.128:60254 (size: 3.5 KB, free: 534.5 MB)

15/09/28 16:18:49 INFO spark.SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:861

15/09/28 16:18:49 INFO scheduler.DAGScheduler: Submitting 2 missing tasks from ResultStage 0 (PythonRDD[2] at count at <stdin>:1)

15/09/28 16:18:49 INFO scheduler.TaskSchedulerImpl: Adding task set 0.0 with 2 tasks

15/09/28 16:18:49 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, 192.168.40.129, ANY, 2157 bytes)

15/09/28 16:18:53 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.40.129:42679 (size: 3.5 KB, free: 534.5 MB)

15/09/28 16:18:57 INFO storage.BlockManagerInfo: Added broadcast_0_piece0 in memory on 192.168.40.129:42679 (size: 14.1 KB, free: 534.5 MB)

相關推薦