[PySpark] Build R&D environment

開發環境

---

基本操做

Ref：Spark的環境搭建

1、啓動集羣

先啓動hadoop，再啓動spark，查看啓動後的狀態：http://node-master:8080

start-all.sh
start-master.sh
start-slaves.sh

關閉順序：stop-master.sh --> stop-slaves.sh --> stop-all.sh。

 

2、命令行

在Spark中存在着多種運行模式，可以使用本地模式運行、可以使用僞分佈式模式運行、使用分佈式模式也存在多種模式如：Spark Mesos模式、Spark YARN模式；

Spark Mesos	官方推薦模式，通用集羣管理，有兩種調度模式：粗粒度模式（Coarse-grained Mode）與細粒度模式（Fine-grained Mode）
Spark YARN	Hadoop YARN資源管理模式
Standalone	簡單模式或稱獨立模式，能夠單獨部署到一個集羣中，無依賴任何其餘資源管理系統。不使用其餘調度工具時會存在單點故障，使用Zookeeper等能夠解決
Local	本地模式，能夠啓動本地一個線程來運行job，能夠啓動N個線程或者使用系統全部核運行job

  

本地運行

1. 本地單線程模式：pyspark --master local
2. 本地多線程模式：pyspark --master local[*] 　　# default
3. 本地多線程模式：pyspark --master local[K]

在程序執行過程當中，只會生成一個SparkSubmit進程。

• • 既是客戶提交任務的 clent進程、
  • 又是Spark的 driver程序、[啓動start-history-server.sh服務查看]
  • 還充當着Spark執行Task的 executor角色。

 

模擬集羣

SparkSubmit 依然充當全能角色，又是Client進程，又是driver程序，還有點資源管理的做用。

提交應用程序時使用 local-cluster[x,y,z] 參數：

x表明要生成的executor數，y和z分別表明每一個executor所擁有的core和memory數。

spark-submit --master local-cluster[2, 3, 1024]
spark-shell  --master local-cluster[2, 3, 1024]

上面這條命令表明會使用2個executor進程，每一個進程分配3個core和1G的內存，來運行應用程序。

 

集羣運行

1. 集羣多線程模式：pyspark --master spark://HOST:PORT（默認7077）

 

3、Yarn 集羣控制

兩種 「部署模式」

1. 開發模式：yarn-client（默認模式），spark driver創建在client上，非集羣上，致使client不能關機。
2. 生產模式： yarn-cluster

# 集羣的位置
hadoop@node-master$ echo $HADOOP_CONF_DIR
/home/hadoop/hadoop/etc/hadoop

你要切換到 yarn-cluster模式 或者 yarn-client模式，很簡單，將咱們以前用於提交spark應用程序的spark-submit腳本，加上--master參數，設置爲yarn-cluster，或yarn-client，便可。若是你沒設置，那麼，就是 standalone模式。

 

提交應用

一個例子，pyspark在提交時，選擇以哪一種模式運行。

#!/usr/bin/env bash
spark-submit \
     --master yarn \
     --deploy-mode cluster \
     --conf spark.shuffle.service.enabled=true \
     --queue xxx \
     --conf spark.dynamicAllocation.enabled=true \
     --conf spark.default.parallelism=1000 \
     --conf spark.sql.shuffle.partitions=1000 \
     --py-files dependencies/dependencies.zip \
     --executor-memory 18g \
     --executor-cores 3 \
     --conf spark.blacklist.enabled=true dependencies/test.py $1 $2 $3 $4 $5 $6 $7 $8

 

4、添加.jar包

運行程序時依賴時使用。

$ cd /usr/local/spark
$ ./bin/pyspark -master local[4] --jars code.jar

 

5、小程序示範

經過pyspark實現wordcount。在編寫spark代碼時，也能夠給SparkContext的setMaster()方法，傳入這個master URL地址；而後咱們的spark做業，就會使用standalone模式鏈接master，並提交做業。

from pyspark import SparkConf, SparkContext

# Init.
conf = SparkConf().setMaster("local").setAppName("My App")
sc = SparkContext(conf = conf)
logFile = "file:///usr/local/spark/README.md"

# Load.
logData = sc.textFile(logFile, 2).cache()

# RDD.
numAs = logData.filter(lambda line: 'a' in line).count()
numBs = logData.filter(lambda line: 'b' in line).count()

print('Lines with a: %s, Lines with b: %s' % (numAs, numBs))

提交執行。

$ /usr/local/spark/bin/spark-submit WordCount.py

 

6、測試程序

關閉調試信息，能夠改成 INFO --> ERROR

hadoop@node-master$ find spark/ -name "*" | xargs grep "log4j.rootCategory=" spark/conf/log4j.properties.template:log4j.rootCategory=INFO, console

 

跟OpenCV同樣，這裏是個大寶藏。 

./bin/spark-submit examples/src/main/python/pi.py

hadoop@node-master$ spark-submit --master spark://node-master:7077 --deploy-mode client examples/src/main/python/

als.py ml/ pi.py streaming/ avro_inputformat.py mllib/ sort.py transitive_closure.py kmeans.py pagerank.py sql/ wordcount.py logistic_regression.py parquet_inputformat.py status_api_demo.py  

 

7、單機與集羣對比

可見，集羣運行要快不少。

 

 

 

集羣操做

Ref: 看了以後再也不迷糊-Spark多種運行模式

真的物理集羣，不是僞集羣。

 

1，測試或實驗性質的本地運行模式 （單機）

2，測試或實驗性質的本地僞集羣運行模式（單機模擬集羣）

3，Spark自帶Cluster Manager的Standalone Client模式（集羣）

 

Ref: 12二、Spark核心編程進階之單獨啓動master和worker腳本 [必要是，單獨自定義配置各個worker]

(1) 爲何咱們有的時候也須要單獨啓動master和worker進程呢?

在單獨啓動兩個進程的時候，是能夠經過命令行參數，爲進程配置一些獨特的參數。
好比說監聽的端口號、web ui的端口號、使用的cpu和內存。
好比你想單獨給某個worker節點配置不一樣的cpu和內存資源的使用限制，那麼就可使用腳本單獨啓動這個worker進程的時候，經過命令行參數來設置。

運行的命令：

 

Worker狀態查看：

   

4，spark自帶cluster manager的standalone cluster模式（集羣） 

hadoop@node-master$ spark-submit --master spark://node-master:7077 --deploy-mode cluster examples/src/main/python/ml/chi_square_test_example.py 

Exception in thread "main" org.apache.spark.SparkException: Cluster deploy mode is currently not supported for python applications on standalone clusters.
    at org.apache.spark.deploy.SparkSubmit.error(SparkSubmit.scala:853)
    at org.apache.spark.deploy.SparkSubmit.prepareSubmitEnvironment(SparkSubmit.scala:273)
    at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:774)
    at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:161)
    at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:184)
    at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
    at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:920)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:929)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

 

5，基於YARN 的Resource Manager的Client模式（集羣）

6，基於YARN 的Resource Manager的Custer模式（集羣） 

如今愈來愈多的場景，都是Spark跑在Hadoop集羣中，因此爲了作到資源可以均衡調度，會使用YARN來作爲Spark的Cluster Manager，來爲Spark的應用程序分配資源。

天然地，須要經過yarn的web ui查看狀態。 

 

 

Spark Mesos 模式

 

/* implement */