我的3个任务想分别在sparkspark 集群搭建中的三台机器上跑怎么办

SPARK 集群安装与简单测试_图文_百度文库
两大类热门资源免费畅读
续费一年阅读会员,立省24元!
SPARK 集群安装与简单测试
上传于||暂无简介
阅读已结束,如果下载本文需要使用1下载券
想免费下载本文?
定制HR最喜欢的简历
下载文档到电脑,查找使用更方便
还剩5页未读,继续阅读
定制HR最喜欢的简历
你可能喜欢微信扫一扫,分享给好友吧。HA下的Spark集群工作原理解密
HA下的Spark集群工作原理解密
第一阶段(月):会从浅入深,基于大量案例实战,深度剖析和讲解,并且会包含完全从企业真实复杂业务需求中抽取出的案例实战。课程会涵盖编程详解、核心编程、和、、、、内核以及源码剖析、性能调优、企业级案例实战等部分
第二阶段(超大规模大数据案例实战):使用了技术生态栈中的、、、、,进行离线计算和实时计算业务模块的开发、数据的关联性分析、用户行为模式和特征的训练与应用、用户网络的社区发现、用户影响力、能量传播、标签传播、标签推理、人群划分、年龄段预测、商品交易时序跳转
本期内容:
1&Spark高可用实战
2&Spark集群工作原理详解
通常资源指、,结构存在单点故障,即容易出现故障,为了应对单点故障,在生产环境下通过做。现通常使用个做,即使有的出现故障,另外两个模式的会由选出,替换故障的,变为,恢复集群状态,通过保存集群的元数据恢复(Zookeeper中包含元数据,所有的、)。
只有模式恢复了才能恢复集群的正常,而在此过程中不能提交作业,切换的过程中不会影响集群的作业的运行,因为程序在运行前,已经向申请过资源,申请过资源后,与分配的进行通信,此过程不需要参与,除非故障。所以,切换的过程不影响作业运行,这是一种粗粒度的管理方式。
粗粒度:一次性分配好资源,之后只需使用或覆用,
优点:在于一次性分配好资源后无需再分配,和的正常交互不受切换的影响;
缺点:较多的时候,只要有一个没有运行完,作业没有运行完,为作 业分配的资源就会闲置起来,
细粒度:需要资源的时候在进行分配,使用完资源后立即释放,缺点:资源无法覆 &&用,任务启动非常慢,且有通信麻烦。
实现高可用实战:
第一步:下载地址:
第二步:解压配置环境
export&/ZOOKEEPER_HOME=/usr/local/zookeeper-3.6.4
并在中添加的目录(安装和,没有关系)
修改:、进到目录下,把修改成
&&& dataDir=.../zookeeper/zookeeper-3.6.4/data(若不修改,重启后会删除
&& &dataLogDir=.../logs(data和是安装目录下创建的目录
&& &指定三台集群
2、zookeeper安装目录下创建和用来存放数据和日志,并在&
&下创建文件并在中添加。
dataDir=/tmp/zookeeper
dataDir=/usr/local/zookeeper-3.4.6/data
dataLogDir=/usr/local/zookeeper-3.4.6/logs
server.0=master:
server.1=worker1:
server.2=worker2:
第三步:拷贝到台上面;将Worker1
中的改为,上的
第四步:分别启动(、、):
.../bin/#&zkServer.sh&start
jps查看进程,多了进程
第五步:让支持:
&vim&spark-env.sh配置的支持信息,
2、配置所有做的机器
&p&&span style=&color:rgb(255,0,0);&&export JAVA_HOME=/root/install/jdk1.7.0_21&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&#export SPARK_MASTER_IP=spark1&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&#export SPARK_MASTER_PORT=7077&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&export SPARK_WORKER_CORES=1&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&export SPARK_WORKER_INSTANCES=1&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&export SPARK_WORKER_MEMORY=1g&/span&&/p&&p&&span style=&color:rgb(255,0,0);&&export SPARK_DAEMON_JAVA_OPTS=&-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=Master:2181,Worker1:;-Dspark.deploy.zookeeper.dir=/spark&&/span&&/p&&p&&strong&其中:&/strong&&/p&&p&-Dspark.deploy.recoveryMode=ZOOKEEPER &span style=&font-family:宋体;&&:说明整个集群状态通过&/span&&span style=&font-family:Times New R&&zookeeper&/span&&span style=&font-family:宋体;&&来维护&/span&&span style=&font-family: Arial, Helvetica, sans-&&也说明整个集群状态恢复也是靠&/span&&span style=&font-family: 'Times New Roman';&&zookeeper&/span&&/p&&p&-Dspark.deploy.zookeeper.url=Master:2181,Worker1:;&span style=&font-family:宋体;&&:说明&/span&&span style=&font-family:Times New R&&zookeeper&/span&&span style=&font-family:宋体;&&做&/span&&span style=&font-family:Times New R&&HA&/span&&span style=&font-family:宋体;&&的机器&/span&&/p&&p&-Dspark.deploy.zookeeper.dir=/spark&&span style=&font-family:宋体;&&:说明存放&/span&&span style=&font-family:Times New R&&Spark&/span&&span style=&font-family:宋体;&&元数据,和&/span&&span style=&font-family:Times New R&&spark&/span&&span style=&font-family:宋体;&&运行状态&/span&&/p&
第六步:启动集群
问题:为什么只在中有进程?
Spark配置文件中指定了,此时需要手动启动、上启动进程:,之后查看的为
状态所以的Workers,所有的Died、所有的Alive
第七步:测试
&./spark-shell&--master&spark://Master:
向去找一个的,只跟的进行交互。
client.AppClient$ClientActor:&Master&haschanged,&new&master&is&at&spark://worker1:7077
关闭机器,恢复集群需要时间,哪台机器恢复快,就会与中的进行交互,即该机器变为,此时界面无法访问,变为,机器变为(切换没有意义,且浪费时间),然后重启集群后,任然变为。
zookeeper做ha,以前一直认为可以从woker选举变为master级别,原来从master(standby)变为master(active),心跳貌似是2秒,默认。
Spark&Runtime&
一般程序提交的时候,当启动后,向进行注册初始化时完成,(程序是在注册的时候完成资源分配)然后由分配资源,程序提交给由的子类进行管理。
DT大数据梦工厂
新浪微博:
微信公众号:
我的热门文章
即使是一小步也想与你分享Spark性能调优
发表于 13:37|
来源程序员电子刊|
作者程序员电子刊
摘要:通常我们对一个系统进行性能优化无怪乎两个步骤——性能监控和参数调整,本文主要分享的也是这两方面内容。
通常我们对一个系统进行性能优化无怪乎两个步骤——性能监控和参数调整,本文主要分享的也是这两方面内容。性能监控工具【Spark监控工具】Spark提供了一些基本的Web监控页面,对于日常监控十分有用。1. Application Web UIhttp://master:4040(默认端口是4040,可以通过spark.ui.port修改)可获得这些信息:(1)stages和tasks调度情况;(2)RDD大小及内存使用;(3)系统环境信息;(4)正在执行的executor信息。2. history server当Spark应用退出后,仍可以获得历史Spark应用的stages和tasks执行信息,便于分析程序不明原因挂掉的情况。配置方法如下:(1)$SPARK_HOME/conf/spark-env.shexport SPARK_HISTORY_OPTS="-Dspark.history.retainedApplications=50Dspark.history.fs.logDirectory=hdfs://hadoop000:8020/directory"说明:spark.history.retainedApplica-tions仅显示最近50个应用spark.history.fs.logDirectory:Spark History Server页面只展示该路径下的信息。(2)$SPARK_HOME/conf/spark-defaults.confspark.eventLog.enabled truespark.eventLog.dir hdfs://hadoop000:8020/directory #应用在运行过程中所有的信息均记录在该属性指定的路径下3. press true(1)HistoryServer启动$SPARK_HOMR/bin/start-histrory-server.sh(2)HistoryServer停止$SPARK_HOMR/bin/stop-histrory-server.sh4. ganglia通过配置ganglia,可以分析集群的使用状况和资源瓶颈,但是默认情况下ganglia是未被打包的,需要在mvn编译时添加-Pspark-ganglia-lgpl,并修改配置文件$SPARK_HOME/conf/metrics.properties。5. Executor logsStandalone模式:$SPARK_HOME/logsYARN模式:在yarn-site.xml文件中配置了YARN日志的存放位置:yarn.nodemanager.log-dirs,或使用命令获取yarn logs -applicationId。【其他监控工具】1. Nmon(/developerworks/aix/library/au-analyze_aix/)Nmon 输入:c:CPU n:网络 m:内存 d:磁盘2. Jmeter(http://jmeter. apache.org/)通常使用Jmeter做系统性能参数的实时展示,JMeter的安装非常简单,从官方网站上下载,解压之后即可使用。运行命令在%JMETER_HOME%/bin下,对于 Windows 用户,直接使用jmeter.bat。启动jmeter:创建测试计划,设置线程组设置循环次数。添加监听器:jp@gc - PerfMon Metrics Collector。设置监听器:监听主机端口及监听内容,例如CPU。启动监听:可以实时获得节点的CPU状态信息,从图4可看出CPU已出现瓶颈。3. Jprofiler(http://www./products/jprofiler/overview.html)JProfiler是一个全功能的Java剖析工具(profiler),专用于分析J2SE和J2EE应用程式。它把CPU、线程和内存的剖析组合在一个强大的应用中。JProfiler的GUI可以更方便地找到性能瓶颈、抓住内存泄漏(memory leaks),并解决多线程的问题。例如分析哪个对象占用的内存比较多;哪个方法占用较大的CPU资源等;我们通常使用Jprofiler来监控Spark应用在local模式下运行时的性能瓶颈和内存泄漏情况。上述几个工具可以直接通过提供的链接了解详细的使用方法。Spark调优【Spark集群并行度】在Spark集群环境下,只有足够高的并行度才能使系统资源得到充分的利用,可以通过修改spark-env.sh来调整Executor的数量和使用资源,Standalone和YARN方式资源的调度管理是不同的。在Standalone模式下:1. 每个节点使用的最大内存数:SPARK_WORKER_INSTANCES*SPARK_WORKER_MEMORY;2. 每个节点的最大并发task数:SPARK_WORKER_INSTANCES*SPARK_WORKER_CORES。在YARN模式下:1. 集群task并行度:SPARK_ EXECUTOR_INSTANCES* SPARK_EXECUTOR_CORES;2. 集群内存总量:(executor个数) * (SPARK_EXECUTOR_MEMORY+ spark.yarn.executor.memoryOverhead)+(SPARK_DRIVER_MEMORY+spark.yarn.driver.memoryOverhead)。重点强调:Spark对Executor和Driver额外添加堆内存大小,Executor端:由spark.yarn.executor.memoryOverhead设置,默认值executorMemory * 0.07与384的最大值。Driver端:由spark.yarn.driver.memoryOverhead设置,默认值driverMemory * 0.07与384的最大值。通过调整上述参数,可以提高集群并行度,让系统同时执行的任务更多,那么对于相同的任务,并行度高了,可以减少轮询次数。举例说明:如果一个stage有100task,并行度为50,那么执行完这次任务,需要轮询两次才能完成,如果并行度为100,那么一次就可以了。但是在资源相同的情况,并行度高了,相应的Executor内存就会减少,所以需要根据实际实况协调内存和core。此外,Spark能够非常有效的支持短时间任务(例如:200ms),因为会对所有的任务复用JVM,这样能减小任务启动的消耗,Standalone模式下,core可以允许1-2倍于物理core的数量进行超配。【Spark任务数量调整】Spark的任务数由stage中的起始的所有RDD的partition之和数量决定,所以需要了解每个RDD的partition的计算方法。以Spark应用从HDFS读取数据为例,HadoopRDD的partition切分方法完全继承于MapReduce中的FileInputFormat,具体的partition数量由HDFS的块大小、mapred.min.split.size的大小、文件的压缩方式等多个因素决定,详情需要参见FileInputFormat的代码。【Spark内存调优】内存优化有三个方面的考虑:对象所占用的内存,访问对象的消耗以及垃圾回收所占用的开销。1. 对象所占内存,优化数据结构Spark 默认使用Java序列化对象,虽然Java对象的访问速度更快,但其占用的空间通常比其内部的属性数据大2-5倍。为了减少内存的使用,减少Java序列化后的额外开销,下面列举一些Spark官网(http://spark.apache.org/docs/latest/tuning.html#tuning-data-structures)提供的方法。(1)使用对象数组以及原始类型(primitive type)数组以替代Java或者Scala集合类(collection class)。fastutil 库为原始数据类型提供了非常方便的集合类,且兼容Java标准类库。(2)尽可能地避免采用含有指针的嵌套数据结构来保存小对象。(3)考虑采用数字ID或者枚举类型以便替代String类型的主键。(4)如果内存少于32GB,设置JVM参数-XX:+UseCom-pressedOops以便将8字节指针修改成4字节。与此同时,在Java 7或者更高版本,设置JVM参数-XX:+UseC-----ompressedStrings以便采用8比特来编码每一个ASCII字符。2. 内存回收(1)获取内存统计信息:优化内存前需要了解集群的内存回收频率、内存回收耗费时间等信息,可以在spark-env.sh中设置SPARK_JAVA_OPTS=“-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps $ SPARK_JAVA_OPTS”来获取每一次内存回收的信息。(2)优化缓存大小:默认情况Spark采用运行内存(spark.executor.memory)的60%来进行RDD缓存。这表明在任务执行期间,有40%的内存可以用来进行对象创建。如果任务运行速度变慢且JVM频繁进行内存回收,或者内存空间不足,那么降低缓存大小设置可以减少内存消耗,可以降低spark.storage.memoryFraction的大小。3. 频繁GC或者OOM针对这种情况,首先要确定现象是发生在Driver端还是在Executor端,然后在分别处理。Driver端:通常由于计算过大的结果集被回收到Driver端导致,需要调大Driver端的内存解决,或者进一步减少结果集的数量。Executor端:(1)以外部数据作为输入的Stage:这类Stage中出现GC通常是因为在Map侧进行map-side-combine时,由于group过多引起的。解决方法可以增加partition的数量(即task的数量)来减少每个task要处理的数据,来减少GC的可能性。(2)以shuffle作为输入的Stage:这类Stage中出现GC的通常原因也是和shuffle有关,常见原因是某一个或多个group的数据过多,也就是所谓的数据倾斜,最简单的办法就是增加shuffle的task数量,比如在SparkSQL中设置SET spark.sql.shuffle.partitions=400,如果调大shuffle的task无法解决问题,说明你的数据倾斜很严重,某一个group的数据远远大于其他的group,需要你在业务逻辑上进行调整,预先针对较大的group做单独处理。【修改序列化】使用Kryo序列化,因为Kryo序列化结果比Java标准序列化更小,更快速。具体方法:spark-default.conf 里设置spark.serializer为org.apache.spark.serializer.KryoSerializer 。参考官方文档(http://spark.apache.org/docs/latest/tuning.html#summary):对于大多数程序而言,采用Kryo框架以及序列化能够解决性能相关的大部分问题。【Spark 磁盘调优】在集群环境下,如果数据分布不均匀,造成节点间任务分布不均匀,也会导致节点间源数据不必要的网络传输,从而大大影响系统性能,那么对于磁盘调优最好先将数据资源分布均匀。除此之外,还可以对源数据做一定的处理:1. 在内存允许范围内,将频繁访问的文件或数据置于内存中;2. 如果磁盘充裕,可以适当增加源数据在HDFS上的备份数以减少网络传输;3. Spark支持多种文件格式及压缩方式,根据不同的应用环境进行合理的选择。如果每次计算只需要其中的某几列,可以使用列式文件格式,以减少磁盘I/O,常用的列式有parquet、rcfile。如果文件过大,将原文件压缩可以减少磁盘I/O,例如:gzip、snappy、lzo。【其他】广播变量(broadcast)当task中需要访问一个Driver端较大的数据时,可以通过使用SparkContext的广播变量来减小每一个任务的大小以及在集群中启动作业的消耗。参考官方文档http://spark.apache.org/docs/latest/tuning.html#broadcasting-large-variables。开启推测机制推测机制后,如果集群中,某一台机器的几个task特别慢,推测机制会将任务分配到其他机器执行,最后Spark会选取最快的作为最终结果。在spark-default.conf 中添加:spark.speculation true推测机制与以下几个参数有关:1. spark.speculation.interval 100:检测周期,单位毫秒;2. spark.speculation.quantile 0.75:完成task的百分比时启动推测;3. spark.speculation.multiplier 1.5:比其他的慢多少倍时启动推测。总结Spark系统的性能调优是一个很复杂的过程,需要对Spark以及Hadoop有足够的知识储备。从业务应用平台(Spark)、存储(HDFS)、操作系统、硬件等多个层面都会对性能产生很大的影响。借助于多种性能监控工具,我们可以很好地了解系统的性能表现,并根据上面介绍的经验进行调整。作者简介:田毅,亚信科技大数据平台部门研发经理,Spark Contributor,北京Spark Meetup发起人,主要关注SparkSQL与Spark Streaming。&<img src="http://ipad-cms.csdn.net/cms/attachment/093eea24117.jpg" 本文选自程序员电子版2015年3月A刊,该期更多文章请查看。2000年创刊至今所有文章目录请查看。欢迎(含iPad版、Android版、PDF版)。
推荐阅读相关主题:
为了更好帮助企业深入了解国内外最新大数据技术,掌握更多行业大数据实践经验,进一步推进大数据技术创新、行业应用和人才培养,-12日,由中国计算机学会(CCF)主办,CCF大数据专家委员会承办,中国科学院计算技术研究所、北京中科天玑科技有限公司及CSDN共同协办的2015中国大数据技术大会(Big Data Technology Conference 2015,BDTC 2015)将在北京新云南皇冠假日酒店隆重举办。
相关热门文章}

我要回帖

更多关于 spark集群 的文章

更多推荐

版权声明:文章内容来源于网络,版权归原作者所有,如有侵权请点击这里与我们联系,我们将及时删除。

点击添加站长微信