码迷,mamicode.com
首页 > 其他好文 > 详细

【Spark深入学习 -12】Spark程序设计与企业级应用案例02

时间:2017-05-07 23:09:16      阅读:181      评论:0      收藏:0      [点我收藏+]

标签:cal   角度   架构   new   hba   关系   aaaaa   example   根据   

----本节内容-------

1.遗留问题答疑

1.1 典型问题解答

1.2 知识点回顾

2.Spark编程基础

2.1 Spark开发四部曲

2.2 RDD典型实例

2.3 非RDD典型实例

3.问题解答

4.参考资料

---------------------

 

技术分享

 

每一次答疑阶段,我都会站在老师的角度去思考一下,如果是我,我应该怎么回答,每每如此,不禁吓出一身冷汗。有些问题看答案确实挺容易的,但当自己作为一个答疑者去思考,可能不一样,因为快速确认一个答案的同时,你得否认很多的东西,脑海里闪过很多的不确定,都要快速否决。董先生无疑是值得敬佩的,这方面体现的很内行,可见内功深厚,他是一个优秀的架构师,也是一个很好的传道者。

1.遗留问题答疑

 

1.1 典型问题解答

1)spark通常和hadoop与yarn一同使用,在学spark过程中补充哪些知识点,可以整理列一下

HDFS 我觉得必须都得清楚,HDFS的东西其实很多,向老师说的文件基本的存储策略,读取操作,基本命令行、API等等;

YARN基本架构,ResoureManger,NodeManger,ApplicationMaster,container等等

其实我想说的是,最好是都过一遍吧,这些都是内功,都是相互依赖的,谁也离不了谁,还有比如Linux知识,Java基本知识,网络基本知识等等,都是实际工作问题中要用到的知识点,我保证,都会用的到。

2)一个partition只会在一个节点上把,老师没有将container的概念么?老师能把partion,container,executor,task再细讲一下么?

临时有事,这里没有听。这个其实挺基础的,但是回答出来比较考验基本功。我把问题衍生一下,包含如下问题:

a).partion的基本概念是什么,partition和block是什么关系?

partition是RDD内部并行计算的一个计算单元,RDD的数据集在逻辑上被划分为多个分片,每一个分片叫做一个partition。block是HDFS存储的最小单元

其实这里还有一个问题,就是Spark为什么要进行分区,出于什么原因要搞一个分区的概念?,我觉得对partiion的理解非常有帮助,答案请参考博文《Spark你妈喊你回家吃饭-08 说说Spark分区原理及优化方法》

b).container的概念是什么,yarn是如何基于container进行资源分配的?

c).Spark应用程序的执行过程是什么?

还有两个问题都是任务提交模式问题,了解spark程序运行的各种模式基本都能解答,这几个问题在前面其实都已经有很明确的讲解过。

 

1.2 知识点回顾

1)核心概念RDD

弹性分布式数据集,把握以下几点

· RDD是什么数据集,他是一个描述数据在哪里,对数据做什么操作,以及操作之间的依赖关系的一个数据集

· 为什么是弹性,主要是说他的存储,既可以在内存,也可以在磁盘

· 分布式:分布在集群上

· 自动重构:失效够可以自动重构

技术分享

 

2) 程序架构

application = 1个driver +多个executor

Excecutor = 多个task+cache

搞清楚driver做了什么,Executor做了什么,Task又做了什么,如何配合

技术分享

 

3)Yarn分布式模式

a.client发送资源申请请求

b.RM发送通知NodeManger要调用资源,

c.NodeManger启动AppAplicationMaster

d.AppAplicationMaster通知nodeManager启动各个Executor

e.nodeManager启动Executor

f.nodeManager向Driver回报实时执行情况,也会告知AppAplicationMaster

技术分享

 

 

2.Spark编程基础

2.1 Spark开发四部曲

每一个spark程序都有一个main,我们称之为driver,driver将程序分解成多个task, task分发到多个executor,从而完成并行化。Spark程序开发的四部曲总结起来如下:

· 创建SparkContext对象

封装了spark执行环境信息

· 创建RDD

可从scala结合或hadoop数据集上创建

· 在RDD上执行转换和action

spark提供了多种转换和action函数

· 返回结果

保存到HDFS中,或者直接打印出来

 

2.2 RDD典型实例

启动spark shell:

bin/spark-shell --master spark://master01:7077 --driver-class-path/usr/local/tdr_hadoop/spark/spark-1.6.0-bin-hadoop2.6/lib/mysql-connector-java-5.1.40-bin.jar

 

1) 设置conf

val conf=new org.apache.spark.SparkConf().setAppName("xx");

conf.set("spark.app.name","test");

conf.set("spark.driver.allowMultipleContexts","true");

val sc= new org.apache.spark.SparkContext(conf);

#读取hdfs上的文件,如果你在hdfs-site.xml中配置hdfs集群的话

val a=sc.textFile("/tmp/test/core-site.xml");

a.collect().foreach(println);

#读取hdfs上的文件

val a = sc.textFile("hdfs:///tmp/test/core-site.xml");

a.collect().foreach(println);

#读取hdfs上的文件 ,这里的端口是9000

vala=sc.textFile("hdfs://master02:9000/tmp/test/core-site.xml");

a.collect().foreach(println);

 

#读取本地文件,这里要注意,driver是在哪里运行,如果driver不在本地运行,执行会报错,找不到文件哦

val a=sc.textFile("file:///data/aa.txt");

a.collect().foreach(println);

 

报错1:netty是spark通信框架,通信超时了。所以产生问题。把::1也就是ip6先注释掉。但是还是没有解决问题,后来把master HA给回退过去了,又恢复了,可能是HA的配置没有配好导致

技术分享

 

报错2:不能出现多个sc

设置参数:conf.set("spark.driver.allowMultipleContexts","true");

 

2)创建RDD

· 从HDFS中读取数据

inputRdd=sc.textFile("hdfs://master01:8020/xx/xx");

· 从本地读取数据

inputRdd=sc.textFile("file://data/input")

· 从Hbase读取数据

技术分享

 

· 从自定义文件格式读取数据

 

技术分享

 

 

2.3 RDD典型实例

1.回忆经典概念

再次提一下RDD的transformation与Action,现在假设你去面试,面试官问你,简单说说你对Transformation和Action的理解,我个人觉得应该回答以下几个知识点,可能你有很多要说的,但是要整理好思路,一个个知识点回答。

1).先说概念

先说RDD概念,RDD弹性分布式数据集。它记录了2类东西一个是数据,一个是操作。数据,RDD记录了数据是保存在内存还是磁盘,而且能控制数据分区。操作,RDD记录了数据之上的各种操作,以及操作之间的依赖关系。最后说一下特点,RDD具有容错,并行,是spark的核心概念。

接着引入Transformation,Transformation是根据特定规则将一个RDD变换为另一个RDD,记录了RDD的演变过程,最后结果是RDD。

Action:将数据聚集起来执行实际的操作,触发所有job,并且保存计算结果,最后结果是数据。

2).后说联系

Transformation记录RDD之间的转换关系,为Action的触发记录RDD逻辑关系。

3).最后说区别

·Transformation不触发job,返回RDD。

·Action触发job,执行计算,返回数据。

 

2.典型例子

1)Transformation例子

例子1:

------------------------

val nums =sc.parallelize(List(1,2,3),3);

nums.collect();

//返回前k个

nums.take(2);

nums.count();

//累加求和

nums.reduce(_+_);

//写数据到hdfs

nums.saveAsTextFile("/tmp/dxc/week2/output1");

import org.apache.spark.rdd._;

nums.saveAsSequenceFile("/tmp/dxc/week2/output2");

nums.saveAsSequenceFile("/tmp/dxc/week2/output2");

//读取sequenceFile格式文件

lines = sc.sequenceFile("/tmp/dxc/week2/output2/") ;

---------

说明:

collection:转为数组,单机概念,保存到driver里面,非常危险,如果非常大

10G数据,2g内存,很可能打爆driver的内存,慎重

 

技术分享

 

从这里可以看出,启动了一个Executor,Executor上面起了2个task,因为是指定了2个分区,分区的个数决定了task的个数。

 

val nums =sc.parallelize(List(1,2,3),3);

nums.collect();

技术分享

 

这里指定3个分区,启动了3个task。

技术分享

 

保存到hdfs也是3个part,如果指定分区为2,那就保存为2个数据块。

 

------------------------

例子2:

--------

valpets=sc.parallelize([("cat",1),("dog",2),("dog",2)]);

pets.reduceByKey(_+_);

pets.groupByKey();

pets.sortByKey();

pets.collect();

---------

说明:reduceByKey自动在本地进行combine

------------------------

例子3:

val line =sc.textFile("/tmp/test/core-site.xml");

val count=line.flatMap(line=>line.split(" ")).map(word=>(word,1)).reduceByKey(_+_);

count.collect();

line.foreach(println);

技术分享

 

 

例子4:

val left = sc.parallelize(List(("spark",1),("hadoop",1),("storm",1)))

val right=sc.parallelize(List(("scala",1),("hadoop",1),("spark",1)))

val result = left.join(right);

val result = left.cogroup(right);

result.collect();

result.foreach(println);

---------

说明:

cogroup:保证key唯一,key只占一行,可用做一个笛卡尔积,cogroup结果如下

(scala,(CompactBuffer(),CompactBuffer(1)))

(spark,(CompactBuffer(1),CompactBuffer(1)))

(hadoop,(CompactBuffer(1),CompactBuffer(1)))

(storm,(CompactBuffer(1),CompactBuffer()))

------------------------

还有其他的RDD操作可以参考之前写的博客。

2.3 非RDD典型实例

1.Accumulator

Accumulator是spark提供的累加器,顾名思义,该变量只能够增加。 只有driver能获取到Accumulator的值(使用value方法),Task只能对其做增加操作(使用 +=)。你也可以在为Accumulator命名(不支持Python),这样就会在spark web ui中显示,可以帮助你了解程序运行的情况。用于监控和调试,记录符合某类特征的数据数据等。

-----------------

//在driver中定义

val accum = sc.accumulator(0, "Example Accumulator") 

//在task中进行累加 

sc.parallelize(1 to10,5).foreach(x=> accum += 1) 

//在driver中输出 accum.value 

//结果将返回10 res: 10

 

技术分享

 

 

---

说明:

在web ui中可以看到Accumulators在task进行累加的具体情况,driver将accumulator收集过来汇总

 

-----------------

3.广播变量

Spark有两种共享变量——累加器、广播变量。广播变量可以让程序高效地向所有工作节点发送一个较大的只读值,以供一个或多个Spark操作使用。高效分发大对象,比如字典map,集合set,每个executor一份,而不是每个task一份.

Spark中分布式执行的代码需要传递到各个Executor的Task上运行。对于一些只读、固定的数据(比如从DB中读出的数据),每次都需要Driver广播到各个Task上,这样效率低下。广播变量允许将变量只广播(提前广播)给各个Executor。该Executor上的各个Task再从所在节点的BlockManager获取变量,而不是从Driver获取变量,从而提升了效率。

一个Executor只需要在第一个Task启动时,获得一份Broadcast数据,之后的Task都从本节点的BlockManager中获取相关数据。

技术分享

 

 

场景1:

val data = Set(1,2,3,4,5,6,7)

val rdd=sc.parallelize(1 to 6,2)

val result =rdd.map(_=>data.size)

result.collect();

场景2:

val data = Set(1,2,3,4,5,6,7)

val bddata=sc.broadcast(data)

val rdd=sc.parallelize(1 to 6,2)

val result =rdd.map(_=>data.size)

result.collect();

区别是:data数值的获取方式,场景1 executor 每次都要从driver那里获取,要和交互7次,而场景2使用广播变量,将data分发到executor,那么driver和executor只需要交互一次。

4.cache的使用

val data=sc.textFile("/tmp/tbMonitordataResultHbase");

val a = data.count

println(a);

 

 

 

执行了29秒

 

 技术分享

val data=sc.textFile("/tmp/tbMonitordataResultHbase");

data.cache()

val a = data.count

println(a);

 

技术分享

 

cache了100多M

val a = data.count

println(a);

data.persist(org.apache.spark.storage.StorageLevel. MEMORY_ONLY)

 

 

3.问题解答

1).spark计算时,数据会写入磁盘么,什么条件下写入磁盘

 shuffle的时候都会写到磁盘,带shuffle的时候会写入到磁盘,有哪算子呢?

2).报错日志里的错误是,ERRORCoarsegRainedExcecutorBackend:RECEIVED SIGNAL TERM ,这个可能是什么原因

虚拟内存超了,被yarn的nodemanager杀了,配置这个参数, 如果是1G内存,可以达到1.1G的使用内存

<property>

<name>yarn.nodemanager.vmem-pmem-ratio</name>

<value>10</value>

</property>

 

3).中文乱码问题,出现中文乱码和spark没有关系

4).resourceManager的工作负担重不重,实际部署时候是否有必要单独到一台机器上?

  机器的规模,50或者100,可以nodemanager合并

5).hbase预分区个数和spark过程中reduce个数相同么

设置,map阶段和region一样,reduce取决于task

 

 

 

4.参考资料

1.董西成ppt

【Spark深入学习 -12】Spark程序设计与企业级应用案例02

标签:cal   角度   架构   new   hba   关系   aaaaa   example   根据   

原文地址:http://www.cnblogs.com/licheng/p/6822416.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!