[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 */