1)前提是定位数据倾斜,是OOM了,还是任务执行缓慢,看日志,看WebUI
2)解决方法,有多个方面
· 避免不必要的shuffle,如使用广播小表的方式,将reduce-side-join提升为map-side-join
·分拆发生数据倾斜的记录,分成几个部分进行,然后合并join后的结果
·改变并行度,可能并行度太少了,导致个别task数据压力大
·两阶段聚合,先局部聚合,再全局聚合
·自定义paritioner,分散key的分布,使其更加均匀
详细解决方案参考博文《Spark数据倾斜优化方法》
15.RDD创建有哪几种方式?
1).使用程序中的集合创建rdd
2).使用本地文件系统创建rdd
3).使用hdfs创建rdd,
4).基于数据库db创建rdd
5).基于Nosql创建rdd,如hbase
6).基于s3创建rdd,
7).基于数据流,如socket创建rdd
如果只回答了前面三种,是不够的,只能说明你的水平还是入门级的,实践过程中有很多种创建方式。
16.Spark并行度怎么设置比较合适
答:spark并行度,每个core承载2~4个partition,如,32个core,那么64~128之间的并行度,也就是
设置64~128个partion,并行读和数据规模无关,只和内存使用量和cpu使用
时间有关
17.Spark中数据的位置是被谁管理的?
答:每个数据分片都对应具体物理位置,数据的位置是被blockManager,无论
数据是在磁盘,内存还是tacyan,都是由blockManager管理
18.Spark的数据本地性有哪几种?
答:Spark中的数据本地性有三种:
a.PROCESS_LOCAL是指读取缓存在本地节点的数据
b.NODE_LOCAL是指读取本地节点硬盘数据
c.ANY是指读取非本地节点数据
通常读取数据PROCESS_LOCAL>NODE_LOCAL>ANY,尽量使数据以PROCESS_LOCAL或NODE_LOCAL方式读取。其中PROCESS_LOCAL还和cache有关,如果RDD经常用的话将该RDD cache到内存中,注意,由于cache是lazy的,所以必须通过一个action的触发,才能真正的将该RDD cache到内存中。
19.rdd有几种操作类型?
1)transformation,rdd由一种转为另一种rdd
2)action,
3)cronroller,crontroller是控制算子,cache,persist,对性能和效率的有很好的支持
三种类型,不要回答只有2中操作
19.rdd有几种操作类型?
1)transformation,rdd由一种转为另一种rdd
2)action,
3)cronroller,crontroller是控制算子,cache,persist,对性能和效率的有很好的支持
三种类型,不要回答只有2中操作
20.Spark如何处理不能被序列化的对象?
将不能序列化的内容封装成object
21.collect功能是什么,其底层是怎么实现的?
答:driver通过collect把集群中各个节点的内容收集过来汇总成结果,collect返回结果是Array类型的,collect把各个节点上的数据抓过来,抓过来数据是Array型,collect对Array抓过来的结果进行合并,合并后Array中只有一个元素,是tuple类型(KV类型的)的。
22.Spaek程序执行,有时候默认为什么会产生很多task,怎么修改默认task执行个数?
答:1)因为输入数据有很多task,尤其是有很多小文件的时候,有多少个输入
block就会有多少个task启动;2)spark中有partition的概念,每个partition都会对应一个task,task越多,在处理大规模数据的时候,就会越有效率。不过task并不是越多越好,如果平时测试,或者数据量没有那么大,则没有必要task数量太多。3)参数可以通过spark_home/conf/spark-default.conf配置文件设置:
spark.sql.shuffle.partitions 50 spark.default.parallelism 10
第一个是针对spark sql的task数量
第二个是非spark sql程序设置生效
23.为什么Spark Application在没有获得足够的资源,job就开始执行了,可能会导致什么什么问题发生?
答:会导致执行该job时候集群资源不足,导致执行job结束也没有分配足够的资源,分配了部分Executor,该job就开始执行task,应该是task的调度线程和Executor资源申请是异步的;如果想等待申请完所有的资源再执行job的:需要将spark.scheduler.maxRegisteredResourcesWaitingTime设置的很大;spark.scheduler.minRegisteredResourcesRatio 设置为1,但是应该结合实际考虑
否则很容易出现长时间分配不到资源,job一直不能运行的情况。
24.map与flatMap的区别
map:对RDD每个元素转换,文件中的每一行数据返回一个数组对象
flatMap:对RDD每个元素转换,然后再扁平化
将所有的对象合并为一个对象,文件中的所有行数据仅返回一个数组
对象,会抛弃值为null的值
25.列举你常用的action?
collect,reduce,take,count,saveAsTextFile等
26.Spark为什么要持久化,一般什么场景下要进行persist操作?
为什么要进行持久化?
spark所有复杂一点的算法都会有persist身影,spark默认数据放在内存,spark很多内容都是放在内存的,非常适合高速迭代,1000个步骤
只有第一个输入数据,中间不产生临时数据,但分布式系统风险很高,所以容易出错,就要容错,rdd出错或者分片可以根据血统算出来,如果没有对父rdd进行persist 或者cache的化,就需要重头做。
以下场景会使用persist
1)某个步骤计算非常耗时,需要进行persist持久化
2)计算链条非常长,重新恢复要算很多步骤,很好使,persist
3)checkpoint所在的rdd要持久化persist,
lazy级别,框架发现有checnkpoint,checkpoint时单独触发一个job,需要重算一遍,checkpoint前
要持久化,写个rdd.cache或者rdd.persist,将结果保存起来,再写checkpoint操作,这样执行起来会非常快,不需要重新计算rdd链条了。checkpoint之前一定会进行persist。
4)shuffle之后为什么要persist,shuffle要进性网络传输,风险很大,数据丢失重来,恢复代价很大
5)shuffle之前进行persist,框架默认将数据持久化到磁盘,这个是框架自动做的。
27.为什么要进行序列化
序列化可以减少数据的体积,减少存储空间,高效存储和传输数据,不好的是使用的时候要反序列化,非常消耗CPU
28.介绍一下join操作优化经验?
答:join其实常见的就分为两类: map-side join 和 reduce-side join。当大表和小表join时,用map-side join能显著提高效率。将多份数据进行关联是数据处理过程中非常普遍的用法,不过在分布式计算系统中,这个问题往往会变的非常麻烦,因为框架提供的 join 操作一般会将所有数据根据 key 发送到所有的 reduce 分区中去,也就是 shuffle 的过程。造成大量的网络以及磁盘IO消耗,运行效率极其低下,这个过程一般被称为 reduce-side-join。如果其中有张表较小的话,我们则可以自己实现在 map 端实现数据关联,跳过大量数据进行 shuffle 的过程,运行时间得到大量缩短,根据不同数据可能会有几倍到数十倍的性能提升。
备注:这个题目面试中非常非常大概率见到,务必搜索相关资料掌握,这里抛砖引玉。
29.介绍一下cogroup rdd实现原理,你在什么场景下用过这个rdd?
答:cogroup的函数实现:这个实现根据两个要进行合并的两个RDD操作,生成一个CoGroupedRDD的实例,这个RDD的返回结果是把相同的key中两个RDD分别进行合并操作,最后返回的RDD的value是一个Pair的实例,这个实例包含两个Iterable的值,第一个值表示的是RDD1中相同KEY的值,第二个值表示的是RDD2中相同key的值.由于做cogroup的操作,需要通过partitioner进行重新分区的操作,因此,执行这个流程时,需要执行一次shuffle的操作(如果要进行合并的两个RDD的都已经是shuffle后的rdd,同时他们对应的partitioner相同时,就不需要执行shuffle,),
场景:表关联查询
30.下面这段代码输出结果是什么?
def joinRdd(sc:SparkContext) {
val name= Array(
Tuple2(1,"spark"),
Tuple2(2,"tachyon"),
Tuple2(3,"hadoop")
)
val score= Array(
Tuple2(1,100),
Tuple2(2,90),
Tuple2(3,80)
)
val namerdd=sc.parallelize(name);
val scorerdd=sc.parallelize(score);
val result = namerdd.join(scorerdd);
result .collect.foreach(println);
}
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
答案:
(1,(Spark,100))
(2,(tachyon,90))
(3,(hadoop,80))
【Spark面试2000题71-100】Spark core面试篇03
Spark Core是Spark的基石,有很多知识点,面试题集的知识点比较跳跃和分散,建议系统学习了Spark知识再看面试题集。今天继续放送最新整理和设计的《Spark面试2000题》题集,仅供参考学习。本篇博文属于梅峰谷原创,转载请注明出处,如果您觉得对您有帮助,请不要吝啬点赞,你的赞,是志愿者们坚持的动力,是早日做出2000道高质量Spark面试题的动力,如有不准确的地方,请留言说明。
一、面试30题(第71-100题)
1.Spark使用parquet文件存储格式能带来哪些好处?
1) 如果说HDFS 是大数据时代分布式文件系统首选标准,那么parquet则是整个大数据时代文件存储格式实时首选标准
2) 速度更快:从使用spark sql操作普通文件CSV和parquet文件速度对比上看,绝大多数情况
会比使用csv等普通文件速度提升10倍左右,在一些普通文件系统无法在spark上成功运行的情况
下,使用parquet很多时候可以成功运行
3) parquet的压缩技术非常稳定出色,在spark sql中对压缩技术的处理可能无法正常的完成工作
(例如会导致lost task,lost executor)但是此时如果使用parquet就可以正常的完成
4) 极大的减少磁盘I/o,通常情况下能够减少75%的存储空间,由此可以极大的减少spark sql处理
数据的时候的数据输入内容,尤其是在spark1.6x中有个下推过滤器在一些情况下可以极大的
减少磁盘的IO和内存的占用,(下推过滤器)
5) spark 1.6x parquet方式极大的提升了扫描的吞吐量,极大提高了数据的查找速度spark1.6和spark1.5x相比而言,提升了大约1倍的速度,在spark1.6X中,操作parquet时候cpu也进行了极大的优化,有效的降低了cpu
6) 采用parquet可以极大的优化spark的调度和执行。我们测试spark如果用parquet可以有效的减少stage的执行消耗,同时可以优化执行路径
2.Executor之间如何共享数据?
答:基于hdfs或者基于tachyon
3.Spark累加器有哪些特点?
1)累加器在全局唯一的,只增不减,记录全局集群的唯一状态
2)在exe中修改它,在driver读取
3)executor级别共享的,广播变量是task级别的共享
两个application不可以共享累加器,但是同一个app不同的job可以共享
4.如何在一个不确定的数据规模的范围内进行排序?
为了提高效率,要划分划分,划分的范围并且是有序的
要么有序,要么降序?
水塘抽样:目的是从一个集合中选取,集合非常答,适合内存
无法容纳数据的时候使用
从N中抽取出K个,N是随机数
5.spark hashParitioner的弊端是什么?
答:HashPartitioner分区的原理很简单,对于给定的key,计算其hashCode,并除于分区的个数取余,如果余数小于0,则用余数+分区的个数,最后返回的值就是这个key所属的分区ID;弊端是数据不均匀,容易导致数据倾斜,极端情况下某几个分区会拥有rdd的所有数据
6.RangePartitioner分区的原理?
答:RangePartitioner分区则尽量保证每个分区中数据量的均匀,而且分区与分区之间是有序的,也就是说一个分区中的元素肯定都是比另一个分区内的元素小或者大;但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。其原理是水塘抽样。可以参考这篇博文
https://www.iteblog.com/archives/1522.html
7.介绍parition和block有什么关联关系?
答:1)hdfs中的block是分布式存储的最小单元,等分,可设置冗余,这样设计有一部分磁盘空间的浪费,但是整齐的block大小,便于快速找到、读取对应的内容;2)Spark中的partion是弹性分布式数据集RDD的最小单元,RDD是由分布在各个节点上的partion组成的。partion是指的spark在计算过程中,生成的数据在计算空间内最小单元,同一份数据(RDD)的partion大小不一,数量不定,是根据application里的算子和最初读入的数据分块数量决定;3)block位于存储空间、partion位于计算空间,block的大小是固定的、partion大小是不固定的,是从2个不同的角度去看数据。
8.Spark应用程序的执行过程是什么?
1)构建Spark Application的运行环境(启动SparkContext),SparkContext向资源管理器(可以是Standalone、Mesos或YARN)注册并申请运行Executor资源;
2).资源管理器分配Executor资源并启动StandaloneExecutorBackend,Executor运行情况将随着心跳发送到资源管理器上;
3).SparkContext构建成DAG图,将DAG图分解成Stage,并把Taskset发送给Task Scheduler。Executor向SparkContext申请Task,Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor。
4).Task在Executor上运行,运行完毕释放所有资源。
9.hbase预分区个数和spark过程中的reduce个数相同么
答:和spark的map个数相同,reduce个数如果没有设置和reduce前的map数相同。
10.如何理解Standalone模式下,Spark资源分配是粗粒度的?
答:spark默认情况下资源分配是粗粒度的,也就是说程序在提交时就分配好资源,后面执行的时候
使用分配好的资源,除非资源出现了故障才会重新分配。比如Spark shell启动,已提交,一注册,哪怕没有任务,worker都会分配资源给executor。
11.Spark如何自定义partitioner分区器?
答:1)spark默认实现了HashPartitioner和RangePartitioner两种分区策略,我们也可以自己扩展分区策略,自定义分区器的时候继承org.apache.spark.Partitioner类,实现类中的三个方法
def numPartitions: Int:这个方法需要返回你想要创建分区的个数;
def getPartition(key: Any): Int:这个函数需要对输入的key做计算,然后返回该key的分区ID,范围一定是0到numPartitions-1;
equals():这个是Java标准的判断相等的函数,之所以要求用户实现这个函数是因为Spark内部会比较两个RDD的分区是否一样。
2)使用,调用parttionBy方法中传入自定义分区对象
参考:http://blog.csdn.net/high2011/article/details/68491115
12.spark中task有几种类型?
答:2种类型:1)result task类型,最后一个task,2是shuffleMapTask类型,除了最后一个task都是
13.union操作是产生宽依赖还是窄依赖?
答:窄依赖
14.rangePartioner分区器特点?
答:rangePartioner尽量保证每个分区中数据量的均匀,而且分区与分区之间是有序的,一个分区中的元素肯定都是比另一个分区内的元素小或者大;但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。RangePartitioner作用:将一定范围内的数映射到某一个分区内,在实现中,分界的算法尤为重要。算法对应的函数是rangeBounds
15.什么是二次排序,你是如何用spark实现二次排序的?(互联网公司常面)
答:就是考虑2个维度的排序,key相同的情况下如何排序,参考博文:http://blog.csdn.net/sundujing/article/details/51399606
16.如何使用Spark解决TopN问题?(互联网公司常面)
答:常见的面试题,参考博文:http://www.cnblogs.com/yurunmiao/p/4898672.html
17.如何使用Spark解决分组排序问题?(互联网公司常面)
组织数据形式:
aa 11
bb 11
cc 34
aa 22
bb 67
cc 29
aa 36
bb 33
cc 30
aa 42
bb 44
cc 49
需求:
1、对上述数据按key值进行分组
2、对分组后的值进行排序
3、截取分组后值得top 3位以key-value形式返回结果
代码如下
val groupTopNRdd = sc.textFile("hdfs://db02:8020/user/hadoop/groupsorttop/groupsorttop.data")
groupTopNRdd.map(_.split(" ")).map(x => (x(0),x(1))).groupByKey().map(
x => {
val xx = x._1
val yy = x._2
(xx,yy.toList.sorted.reverse.take(3))
}
).collect
18.窄依赖父RDD的partition和子RDD的parition是不是都是一对一的关系?
答:不一定,除了一对一的窄依赖,还包含一对固定个数的窄依赖(就是对父RDD的依赖的Partition的数量不会随着RDD数量规模的改变而改变),比如join操作的每个partiion仅仅和已知的partition进行join,这个join操作是窄依赖,依赖固定数量的父rdd,因为是确定的partition关系
19.Hadoop中,Mapreduce操作的mapper和reducer阶段相当于spark中的哪几个算子?
答:相当于spark中的map算子和reduceByKey算子,当然还是有点区别的,MR会自动进行排序的,spark要看你用的是什么partitioner
20.什么是shuffle,以及为什么需要shuffle?
shuffle中文翻译为洗牌,需要shuffle的原因是:某种具有共同特征的数据汇聚到一个计算节点上进行计算
21.不需要排序的hash shuffle是否一定比需要排序的sort shuffle速度快?
答:不一定!!当数据规模小,Hash shuffle快于Sorted Shuffle数据规模大的时候;当数据量大,sorted Shuffle会比Hash shuffle快很多,因为数量大的有很多小文件,不均匀,甚至出现数据倾斜,消耗内存大,1.x之前spark使用hash,适合处理中小规模,1.x之后,增加了Sorted shuffle,Spark更能胜任大规模处理了。
22.Spark中的HashShufle的有哪些不足?
答:1)shuffle产生海量的小文件在磁盘上,此时会产生大量耗时的、低效的IO操作;2).容易导致内存不够用,由于内存需要保存海量的文件操作句柄和临时缓存信息,如果数据处理规模比较大的化,容易出现OOM;3)容易出现数据倾斜,导致OOM
23.conslidate是如何优化Hash shuffle时在map端产生的小文件?
答:1)conslidate为了解决Hash Shuffle同时打开过多文件导致Writer handler内存使用过大以及产生过多文件导致大量的随机读写带来的低效磁盘IO;2)conslidate根据CPU的个数来决定每个task shuffle map端产生多少个文件,假设原来有10个task,100个reduce,每个CPU有10个CPU
那么使用hash shuffle会产生10*100=1000个文件,conslidate产生10*10=100个文件
备注:conslidate部分减少了文件和文件句柄,并行读很高的情况下(task很多时)还是会很多文件
24.Sort-basesd shuffle产生多少个临时文件
答:2*Map阶段所有的task数量,Mapper阶段中并行的Partition的总数量,其实就是Mapper端task
25.Sort-based shuffle的缺陷?
1) 如果mapper中task的数量过大,依旧会产生很多小文件,此时在shuffle传递数据的过程中reducer段,reduce会需要同时大量的记录进行反序列化,导致大量的内存消耗和GC的巨大负担,造成系统缓慢甚至崩溃
2)如果需要在分片内也进行排序,此时需要进行mapper段和reducer段的两次排序
26.Spark shell启动时会启动derby?
答: spark shell启动会启动spark sql,spark sql默认使用derby保存元数据,但是尽量不要用derby,它是单实例,不利于开发。会在本地生成一个文件metastore_db,如果启动报错,就把那个文件给删了 ,derby数据库是单实例,不能支持多个用户同时操作,尽量避免使用
27.spark.default.parallelism这个参数有什么意义,实际生产中如何设置?
答:1)参数用于设置每个stage的默认task数量。这个参数极为重要,如果不设置可能会直接影响你的Spark作业性能;2)很多人都不会设置这个参数,会使得集群非常低效,你的cpu,内存再多,如果task始终为1,那也是浪费,spark官网建议task个数为CPU的核数*executor的个数的2~3倍。
28.spark.storage.memoryFraction参数的含义,实际生产中如何调优?
答:1)用于设置RDD持久化数据在Executor内存中能占的比例,默认是0.6,,默认Executor 60%的内存,可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略,如果内存不够时,可能数据就不会持久化,或者数据会写入磁盘。2)如果持久化操作比较多,可以提高spark.storage.memoryFraction参数,使得更多的持久化数据保存在内存中,提高数据的读取性能,如果shuffle的操作比较多,有很多的数据读写操作到JVM中,那么应该调小一点,节约出更多的内存给JVM,避免过多的JVM gc发生。在web ui中观察如果发现gc时间很长,可以设置spark.storage.memoryFraction更小一点。
29.spark.shuffle.memoryFraction参数的含义,以及优化经验?
答:1)spark.shuffle.memoryFraction是shuffle调优中 重要参数,shuffle从上一个task拉去数据过来,要在Executor进行聚合操作,聚合操作时使用Executor内存的比例由该参数决定,默认是20%
如果聚合时数据超过了该大小,那么就会spill到磁盘,极大降低性能;2)如果Spark作业中的RDD持久化操作较少,shuffle操作较多时,建议降低持久化操作的内存占比,提高shuffle操作的内存占比比例,避免shuffle过程中数据过多时内存不够用,必须溢写到磁盘上,降低了性能。此外,如果发现作业由于频繁的gc导致运行缓慢,意味着task执行用户代码的内存不够用,那么同样建议调低这个参数的值
30.介绍一下你对Unified Memory Management内存管理模型的理解?
答:Spark中的内存使用分为两部分:执行(execution)与存储(storage)。执行内存主要用于shuffles、joins、sorts和aggregations,存储内存则用于缓存或者跨节点的内部数据传输。1.6之前,对于一个Executor,内存都有哪些部分构成:
1)ExecutionMemory。这片内存区域是为了解决 shuffles,joins, sorts and aggregations 过程中为了避免频繁IO需要的buffer。 通过spark.shuffle.memoryFraction(默认 0.2) 配置。
2)StorageMemory。这片内存区域是为了解决 block cache(就是你显示调用dd.cache, rdd.persist等方法), 还有就是broadcasts,以及task results的存储。可以通过参数 spark.storage.memoryFraction(默认0.6)。设置
3)OtherMemory。给系统预留的,因为程序本身运行也是需要内存的。 (默认为0.2).
传统内存管理的不足:
1).Shuffle占用内存0.2*0.8,内存分配这么少,可能会将数据spill到磁盘,频繁的磁盘IO是很大的负担,Storage内存占用0.6,主要是为了迭代处理。传统的Spark内存分配对操作人的要求非常高。(Shuffle分配内存:ShuffleMemoryManager, TaskMemoryManager,ExecutorMemoryManager)一个Task获得全部的Execution的Memory,其他Task过来就没有内存了,只能等待。
2).默认情况下,Task在线程中可能会占满整个内存,分片数据特别大的情况下就会出现这种情况,其他Task没有内存了,剩下的cores就空闲了,这是巨大的浪费。这也是人为操作的不当造成的。
3).MEMORY_AND_DISK_SER的storage方式,获得RDD的数据是一条条获取,iterator的方式。如果内存不够(spark.storage.unrollFraction),unroll的读取数据过程,就是看内存是否足够,如果足够,就下一条。unroll的space是从Storage的内存空间中获得的。unroll的方式失败,就会直接放磁盘。
4). 默认情况下,Task在spill到磁盘之前,会将部分数据存放到内存上,如果获取不到内存,就不会执行。永无止境的等待,消耗CPU和内存。
在此基础上,Spark提出了UnifiedMemoryManager,不再分ExecutionMemory和Storage Memory,实际上还是分的,只不过是Execution Memory访问Storage Memory,Storage Memory也可以访问Execution Memory,如果内存不够,就会去借。
【Spark面试2000题101-130】Spark on Yarn面试篇04
本篇题集主要是Spark on Yarn相关的面试题,主要涉及Spark on Yarn、Yarn、Mapreduce相关面试题。
一、面试题30题
1.MRV1有哪些不足?
1)可扩展性(对于变化的应付能力)
a) JobTracker内存中保存用户作业的信息
b) JobTracker使用的是粗粒度的锁
2)可靠性和可用性
a) JobTracker失效会多事集群中所有的运行作业,用户需手动重新提交和恢复工作流
3)对不同编程模型的支持
HadoopV1以MapReduce为中心的设计虽然能支持广泛的用例,但是并不适合所有大型计算,如storm,spark
2.描述Yarn执行一个任务的过程?
1)客户端client向ResouceManager提交Application,ResouceManager接受Application
并根据集群资源状况选取一个node来启动Application的任务调度器driver(ApplicationMaster)
2)ResouceManager找到那个node,命令其该node上的nodeManager来启动一个新的
JVM进程运行程序的driver(ApplicationMaster)部分,driver(ApplicationMaster)启动时会首先向ResourceManager注册,说明由自己来负责当前程序的运行
3)driver(ApplicationMaster)开始下载相关jar包等各种资源,基于下载的jar等信息决定向ResourceManager申请具体的资源内容。
4)ResouceManager接受到driver(ApplicationMaster)提出的申请后,会最大化的满足
资源分配请求,并发送资源的元数据信息给driver(ApplicationMaster);
5)driver(ApplicationMaster)收到发过来的资源元数据信息后会根据元数据信息发指令给具体
机器上的NodeManager,让其启动具体的container。
6)NodeManager收到driver发来的指令,启动container,container启动后必须向driver(ApplicationMaster)注册。
7)driver(ApplicationMaster)收到container的注册,开始进行任务的调度和计算,直到
任务完成。
补充:如果ResourceManager第一次没有能够满足driver(ApplicationMaster)的资源请求
,后续发现有空闲的资源,会主动向driver(ApplicationMaster)发送可用资源的元数据信息
以提供更多的资源用于当前程序的运行。
3.Yarn中的container是由谁负责销毁的,在Hadoop Mapreduce中container可以复用么?
答:ApplicationMaster负责销毁,在Hadoop Mapreduce不可以复用,在spark on yarn程序container可以复用
4.提交任务时,如何指定Spark Application的运行模式?
1)cluster模式:./spark-submit –class xx.xx.xx –master yarn –deploy-mode cluster xx.jar
2) client模式:./spark-submit –class xx.xx.xx –master yarn –deploy-mode client xx.jar
5.不启动Spark集群Master和work服务,可不可以运行Spark程序?
答:可以,只要资源管理器第三方管理就可以,如由yarn管理,spark集群不启动也可以使用spark;spark集群启动的是work和master,这个其实就是资源管理框架,yarn中的resourceManager相当于master,NodeManager相当于worker,做计算是Executor,和spark集群的work和manager可以没关系,归根接底还是JVM的运行,只要所在的JVM上安装了spark就可以。
6.Spark中的4040端口由什么功能?
答:收集Spark作业运行的信息
7.spark on yarn Cluster 模式下,ApplicationMaster和driver是在同一个进程么?
答:是,driver 位于ApplicationMaster进程中。该进程负责申请资源,还负责监控程序、资源的动态情况。
8.如何使用命令查看application运行的日志信息
答:yarn logs -applicationId
9.Spark on Yarn 模式有哪些优点?
1)与其他计算框架共享集群资源(eg.Spark框架与MapReduce框架同时运行,如果不用Yarn进行资源分配,MapReduce分到的内存资源会很少,效率低下);资源按需分配,进而提高集群资源利用等。
2)相较于Spark自带的Standalone模式,Yarn的资源分配更加细致
3)Application部署简化,例如Spark,Storm等多种框架的应用由客户端提交后,由Yarn负责资源的管理和调度,利用Container作为资源隔离的单位,以它为单位去使用内存,cpu等。
4)Yarn通过队列的方式,管理同时运行在Yarn集群中的多个服务,可根据不同类型的应用程序负载情况,调整对应的资源使用量,实现资源弹性管理。
10.谈谈你对container的理解?
1)Container作为资源分配和调度的基本单位,其中封装了的资源如内存,CPU,磁盘,网络带宽等。 目前yarn仅仅封装内存和CPU
2)Container由ApplicationMaster向ResourceManager申请的,由ResouceManager中的资源调度器异步分配给ApplicationMaster
3) Container的运行是由ApplicationMaster向资源所在的NodeManager发起的,Container运行时需提供内部执行的任务命令.
11.运行在yarn中Application有几种类型的container?
1) 运行ApplicationMaster的Container:这是由ResourceManager(向内部的资源调度器)申请和启动的,用户提交应用程序时,可指定唯一的ApplicationMaster所需的资源;
2) 运行各类任务的Container:这是由ApplicationMaster向ResourceManager申请的,并由ApplicationMaster与NodeManager通信以启动之。
12.Spark on Yarn架构是怎么样的?(要会画哦,这个图)
Yarn提到的App Master可以理解为Spark中Standalone模式中的driver。Container中运行着Executor,在Executor中以多线程并行的方式运行Task。运行过程和第二题相似。
13.Executor启动时,资源通过哪几个参数指定?
1)num-executors是executor的数量
2)executor-memory 是每个executor使用的内存
3)executor-cores 是每个executor分配的CPU
14.为什么会产生yarn,解决了什么问题,有什么优势?
1)为什么产生yarn,针对MRV1的各种缺陷提出来的资源管理框架
2)解决了什么问题,有什么优势,参考这篇博文:http://www.aboutyun.com/forum.php?mod=viewthread&tid=6785
15.Mapreduce的执行过程?
阶段1:input/map/partition/sort/spill
阶段2:mapper端merge
阶段3:reducer端merge/reduce/output
详细过程参考这个http://www.cnblogs.com/hipercomer/p/4516581.html
16.一个task的map数量由谁来决定?
一般情况下,在输入源是文件的时候,一个task的map数量由splitSize来决定的,那么splitSize是由以下几个来决定的
goalSize = totalSize / mapred.map.tasks
inSize = max {mapred.min.split.size, minSplitSize}
splitSize = max (minSize, min(goalSize, dfs.block.size))
一个task的reduce数量,由partition决定。
17.reduce后输出的数据量有多大?
并不是想知道确切的数据量有多大这个,而是想问你,MR的执行机制,开发完程序,有没有认真评估程序运行效率
1)用于处理redcue任务的资源情况,如果是MRV1的话,分了多少资源给map,多少个reduce
如果是MRV2的话,可以提一下,集群有分了多少内存、CPU给yarn做计算 。
2)结合实际应用场景回答,输入数据有多大,大约多少条记录,做了哪些逻辑操作,输出的时候有多少条记录,执行了多久,reduce执行时候的数据有没有倾斜等
3)再提一下,针对mapReduce做了哪几点优化,速度提升了多久,列举1,2个优化点就可以
18.你的项目提交到job的时候数据量有多大?
答:1)回答出数据是什么格式,有没有采用什么压缩,采用了压缩的话,压缩比大概是多少;2)文件大概多大:大概起了多少个map,起了多少个reduce,map阶段读取了多少数据,reduce阶段读取了多少数据,程序大约执行了多久,3)集群什么规模,集群有多少节点,多少内存,多少CPU核数等。把这些点回答进去,而不是给个数字了事。
19.你们提交的job任务大概有多少个?这些job执行完大概用多少时间?
还是考察你开发完程序有没有认真观察过程序的运行,有没有评估程序运行的效率
20.你们业务数据量多大?有多少行数据?
这个也是看你们有没有实际的经验,对于没有实战的同学,请把回答的侧重点放在MR的运行机制上面,
MR运行效率方面,以及如何优化MR程序(看别人的优化demo,然后在虚拟机上拿demo做一下测试)。
22.如何杀死一个正在运行的job
杀死一个job
MRV1:Hadoop job kill jobid
YARN: yarn application -kill applicationId
23.列出你所知道的调度器,说明其工作原理
a) Fifo schedular 默认的调度器 先进先出
b) Capacity schedular 计算能力调度器 选择占用内存小 优先级高的
c) Fair schedular 调肚脐 公平调度器 所有job 占用相同资源
24.YarnClient模式下,执行Spark SQL报这个错,Exception in thread “Thread-2” java.lang.OutOfMemoryError: PermGen space,但是在Yarn Cluster模式下正常运行,可能是什么原因?
1)原因查询过程中调用的是Hive的获取元数据信息、SQL解析,并且使用Cglib等进行序列化反序列化,中间可能产生较多的class文件,导致JVM中的持久代使用较多
Cluster模式的持久代默认大小是64M,Client模式的持久代默认大小是32M,而Driver端进行SQL处理时,其持久代的使用可能会达到90M,导致OOM溢出,任务失败。
yarn-cluster模式下出现,yarn-client模式运行时倒是正常的,原来在SPARKHOME/bin/spark−class文件中已经设置了持久代大小:JAVAOPTS=”−XX:MaxPermSize=256mSPARKHOME/bin/spark−class文件中已经设置了持久代大小:JAVAOPTS=”−XX:MaxPermSize=256mOUR_JAVA_OPTS”
2)解决方法:在Spark的conf目录中的spark-defaults.conf里,增加对Driver的JVM配置,因为Driver才负责SQL的解析和元数据获取。配置如下:
spark.driver.extraJavaOptions -XX:PermSize=128M -XX:MaxPermSize=256M
25.spark.driver.extraJavaOptions这个参数是什么意思,你们生产环境配了多少?
传递给executors的JVM选项字符串。例如GC设置或者其它日志设置。注意,在这个选项中设置Spark属性或者堆大小是不合法的。Spark属性需要用SparkConf对象或者spark-submit脚本用到的spark-defaults.conf文件设置。堆内存可以通过spark.executor.memory设置
26.导致Executor产生FULL gc 的原因,可能导致什么问题?
答:可能导致Executor僵死问题,海量数据的shuffle和数据倾斜等都可能导致full gc。以shuffle为例,伴随着大量的Shuffle写操作,JVM的新生代不断GC,Eden Space写满了就往Survivor Space写,同时超过一定大小的数据会直接写到老生代,当新生代写满了之后,也会把老的数据搞到老生代,如果老生代空间不足了,就触发FULL GC,还是空间不够,那就OOM错误了,此时线程被Blocked,导致整个Executor处理数据的进程被卡住
27.Combiner 和partition的作用
combine分为map端和reduce端,作用是把同一个key的键值对合并在一起,可以自定义的。combine函数把一个map函数产生的
Hbase 相关试题
Storm 相关试题
大数据相关试题
面试|大数据相关试题-面试篇07
面试系列重新继续发布,下面这个是从网上搜来的,题目都是好题目,答案作为参考是可以的,作为学习素材,仅供大家参考。
1、简答说一下hadoop的map-reduce编程模型
首先map task会从本地文件系统读取数据,转换成key-value形式的键值对集合
使用的是hadoop内置的数据类型,比如longwritable、text等
将键值对集合输入mapper进行业务处理过程,将其转换成需要的key-value在输出
之后会进行一个partition分区操作,默认使用的是hashpartitioner,可以通过重写hashpartitioner的getpartition方法来自定义分区规则
之后会对key进行进行sort排序,grouping分组操作将相同key的value合并分组输出,在这里可以使用自定义的数据类型,重写WritableComparator的Comparator方法来自定义排序规则,重写RawComparator的compara方法来自定义分组规则
之后进行一个combiner归约操作,其实就是一个本地段的reduce预处理,以减小后面shufle和reducer的工作量
reduce task会通过网络将各个数据收集进行reduce处理,最后将数据保存或者显示,结束整个job
2、hadoop的TextInputFormat作用是什么,如何自定义实现
InputFormat会在map操作之前对数据进行两方面的预处理
1是getSplits,返回的是InputSplit数组,对数据进行split分片,每片交给map操作一次
2是getRecordReader,返回的是RecordReader对象,对每个split分片进行转换为key-value键值对格式传递给map
常用的InputFormat是TextInputFormat,使用的是LineRecordReader对每个分片进行键值对的转换,以行偏移量作为键,行内容作为值
自定义类继承InputFormat接口,重写createRecordReader和isSplitable方法
在createRecordReader中可以自定义分隔符
3、hadoop和spark的都是并行计算,那么他们有什么相同和区别
两者都是用mr模型来进行并行计算,hadoop的一个作业称为job,job里面分为map task和reduce task,每个task都是在自己的进程中运行的,当task结束时,进程也会结束
spark用户提交的任务成为application,一个application对应一个sparkcontext,app中存在多个job,每触发一次action操作就会产生一个job
这些job可以并行或串行执行,每个job中有多个stage,stage是shuffle过程中DAGSchaduler通过RDD之间的依赖关系划分job而来的,每个stage里面有多个task,组成taskset有TaskSchaduler分发到各个executor中执行,executor的生命周期是和app一样的,即使没有job运行也是存在的,所以task可以快速启动读取内存进行计算
hadoop的job只有map和reduce操作,表达能力比较欠缺而且在mr过程中会重复的读写hdfs,造成大量的io操作,多个job需要自己管理关系
spark的迭代计算都是在内存中进行的,API中提供了大量的RDD操作如join,groupby等,而且通过DAG图可以实现良好的容错
4、为什么要用flume导入hdfs,hdfs的构架是怎样的
flume可以实时的导入数据到hdfs中,当hdfs上的文件达到一个指定大小的时候会形成一个文件,或者超过指定时间的话也形成一个文件
文件都是存储在datanode上面的,namenode记录着datanode的元数据信息,而namenode的元数据信息是存在内存中的,所以当文件切片很小或者很多的时候会卡死
5、map-reduce程序运行的时候会有什么比较常见的问题
比如说作业中大部分都完成了,但是总有几个reduce一直在运行
这是因为这几个reduce中的处理的数据要远远大于其他的reduce,可能是因为对键值对任务划分的不均匀造成的数据倾斜
解决的方法可以在分区的时候重新定义分区规则对于value数据很多的key可以进行拆分、均匀打散等处理,或者是在map端的combiner中进行数据预处理的操作
6、简单说一下hadoop和spark的shuffle过程
hadoop:map端保存分片数据,通过网络收集到reduce端
spark:spark的shuffle是在DAGSchedular划分Stage的时候产生的,TaskSchedule要分发Stage到各个worker的executor,减少shuffle可以提高性能
7、Hive中存放是什么?
表(数据+元数据)。 存的是和hdfs的映射关系,hive是逻辑上的数据仓库,实际操作的都是hdfs上的文件,HQL就是用sql语法来写的mr程序。
8、Hive与关系型数据库的关系?
没有关系,hive是数据仓库,不能和数据库一样进行实时的CURD操作。
是一次写入多次读取的操作,可以看成是ETL工具。
9、Flume工作机制是什么?
核心概念是agent,里面包括source、chanel和sink三个组件。
source运行在日志收集节点进行日志采集,之后临时存储在chanel中,sink负责将chanel中的数据发送到目的地。
只有成功发送之后chanel中的数据才会被删除。
首先书写flume配置文件,定义agent、source、chanel和sink然后将其组装,执行flume-ng命令。
10、Sqoop工作原理是什么?
hadoop生态圈上的数据传输工具。
可以将关系型数据库的数据导入非结构化的hdfs、hive或者bbase中,也可以将hdfs中的数据导出到关系型数据库或者文本文件中。
使用的是mr程序来执行任务,使用jdbc和关系型数据库进行交互。
import原理:通过指定的分隔符进行数据切分,将分片传入各个map中,在map任务中在每行数据进行写入处理没有reduce。
export原理:根据要操作的表名生成一个java类,并读取其元数据信息和分隔符对非结构化的数据进行匹配,多个map作业同时执行写入关系型数据库
11、Hbase行健列族的概念,物理模型,表的设计原则?
行健:是hbase表自带的,每个行健对应一条数据。
列族:是创建表时指定的,为列的集合,每个列族作为一个文件单独存储,存储的数据都是字节数组,其中的数据可以有很多,通过时间戳来区分。
物理模型:整个hbase表会拆分为多个region,每个region记录着行健的起始点保存在不同的节点上,查询时就是对各个节点的并行查询,当region很大时使用.META表存储各个region的起始点,-ROOT又可以存储.META的起始点。
rowkey的设计原则:各个列簇数据平衡,长度原则、相邻原则,创建表的时候设置表放入regionserver缓存中,避免自动增长和时间,使用字节数组代替string,最大长度64kb,最好16字节以内,按天分表,两个字节散列,四个字节存储时分毫秒。
列族的设计原则:尽可能少(按照列族进行存储,按照region进行读取,不必要的io操作),经常和不经常使用的两类数据放入不同列族中,列族名字尽可能短。
12、Spark Streaming和Storm有何区别?
一个实时毫秒一个准实时亚秒,不过storm的吞吐率比较低。
13、mllib支持的算法?
大体分为四大类,分类、聚类、回归、协同过滤。
14、简答说一下hadoop的map-reduce编程模型?
首先map task会从本地文件系统读取数据,转换成key-value形式的键值对集合。
将键值对集合输入mapper进行业务处理过程,将其转换成需要的key-value在输出。
之后会进行一个partition分区操作,默认使用的是hashpartitioner,可以通过重写hashpartitioner的getpartition方法来自定义分区规则。
之后会对key进行进行sort排序,grouping分组操作将相同key的value合并分组输出。
在这里可以使用自定义的数据类型,重写WritableComparator的Comparator方法来自定义排序规则,重写RawComparator的compara方法来自定义分组规则。
之后进行一个combiner归约操作,其实就是一个本地段的reduce预处理,以减小后面shufle和reducer的工作量。
reduce task会通过网络将各个数据收集进行reduce处理,最后将数据保存或者显示,结
束整个job。
15、Hadoop平台集群配置、环境变量设置?
zookeeper:修改zoo.cfg文件,配置dataDir,和各个zk节点的server地址端口,tickTime心跳时间默认是2000ms,其他超时的时间都是以这个为基础的整数倍,之后再dataDir对应目录下写入myid文件和zoo.cfg中的server相对应。
hadoop:修改
hadoop-env.sh配置java环境变量
core-site.xml配置zk地址,临时目录等
hdfs-site.xml配置nn信息,rpc和http通信地址,nn自动切换、zk连接超时时间等
yarn-site.xml配置resourcemanager地址
mapred-site.xml配置使用yarn
slaves配置节点信息
格式化nn和zk。
hbase:修改
hbase-env.sh配置java环境变量和是否使用自带的zk
hbase-site.xml配置hdfs上数据存放路径,zk地址和通讯超时时间、master节点
regionservers配置各个region节点
zoo.cfg拷贝到conf目录下
spark:
安装Scala
修改spark-env.sh配置环境变量和master和worker节点配置信息
环境变量的设置:直接在/etc/profile中配置安装的路径即可,或者在当前用户的宿主目录下,配置在.bashrc文件中,该文件不用source重新打开shell窗口即可,配置在.bash_profile的话只对当前用户有效。
16、Hadoop性能调优?
调优可以通过系统配置、程序编写和作业调度算法来进行。
hdfs的block.size可以调到128/256(网络很好的情况下,默认为64)
调优的大头:mapred.map.tasks、mapred.reduce.tasks设置mr任务数(默认都是1)
mapred.tasktracker.map.tasks.maximum每台机器上的最大map任务数
mapred.tasktracker.reduce.tasks.maximum每台机器上的最大reduce任务数
mapred.reduce.slowstart.completed.maps配置reduce任务在map任务完成到百分之几的时候开始进入
这个几个参数要看实际节点的情况进行配置,reduce任务是在33%的时候完成copy,要在这之前完成map任务,(map可以提前完成)
mapred.compress.map.output,mapred.output.compress配置压缩项,消耗cpu提升网络和磁盘io
合理利用combiner
注意重用writable对象
17、Hadoop高并发?
首先肯定要保证集群的高可靠性,在高并发的情况下不会挂掉,支撑不住可以通过横向扩展。
datanode挂掉了使用hadoop脚本重新启动。
18、hadoop的TextInputFormat作用是什么,如何自定义实现?
InputFormat会在map操作之前对数据进行两方面的预处理。
1是getSplits,返回的是InputSplit数组,对数据进行split分片,每片交给map操作一次 。
2是getRecordReader,返回的是RecordReader对象,对每个split分片进行转换为key-value键值对格式传递给map。
常用的InputFormat是TextInputFormat,使用的是LineRecordReader对每个分片进行键值对的转换,以行偏移量作为键,行内容作为值。
自定义类继承InputFormat接口,重写createRecordReader和isSplitable方法 。
在createRecordReader中可以自定义分隔符。
19、hadoop和spark的都是并行计算,那么他们有什么相同和区别?
两者都是用mr模型来进行并行计算,hadoop的一个作业称为job,job里面分为map task和reduce task,每个task都是在自己的进程中运行的,当task结束时,进程也会结束。
spark用户提交的任务成为application,一个application对应一个sparkcontext,app中存在多个job,每触发一次action操作就会产生一个job。
这些job可以并行或串行执行,每个job中有多个stage,stage是shuffle过程中DAGSchaduler通过RDD之间的依赖关系划分job而来的,每个stage里面有多个task,组成taskset有TaskSchaduler分发到各个executor中执行,executor的生命周期是和app一样的,即使没有job运行也是存在的,所以task可以快速启动读取内存进行计算。
hadoop的job只有map和reduce操作,表达能力比较欠缺而且在mr过程中会重复的读写hdfs,造成大量的io操作,多个job需要自己管理关系。
spark的迭代计算都是在内存中进行的,API中提供了大量的RDD操作如join,groupby等,而且通过DAG图可以实现良好的容错。
20、为什么要用flume导入hdfs,hdfs的构架是怎样的?
flume可以实时的导入数据到hdfs中,当hdfs上的文件达到一个指定大小的时候会形成一个文件,或者超过指定时间的话也形成一个文件。
文件都是存储在datanode上面的,namenode记录着datanode的元数据信息,而namenode的元数据信息是存在内存中的,所以当文件切片很小或者很多的时候会卡死。
21、map-reduce程序运行的时候会有什么比较常见的问题?
比如说作业中大部分都完成了,但是总有几个reduce一直在运行。
这是因为这几个reduce中的处理的数据要远远大于其他的reduce,可能是因为对键值对任务划分的不均匀造成的数据倾斜。
解决的方法可以在分区的时候重新定义分区规则对于value数据很多的key可以进行拆分、均匀打散等处理,或者是在map端的combiner中进行数据预处理的操作。
16、简单说一下hadoop和spark的shuffle过程?
hadoop:map端保存分片数据,通过网络收集到reduce端。
spark:spark的shuffle是在DAGSchedular划分Stage的时候产生的,TaskSchedule要分发Stage到各个worker的executor。
减少shuffle可以提高性能。
22、RDD机制?
rdd分布式弹性数据集,简单的理解成一种数据结构,是spark框架上的通用货币。
所有算子都是基于rdd来执行的,不同的场景会有不同的rdd实现类,但是都可以进行互相转换。
rdd执行过程中会形成dag图,然后形成lineage保证容错性等。
从物理的角度来看rdd存储的是block和node之间的映射。
18、spark有哪些组件?
(1)master:管理集群和节点,不参与计算。
(2)worker:计算节点,进程本身不参与计算,和master汇报。
(3)Driver:运行程序的main方法,创建spark context对象。
(4)spark context:控制整个application的生命周期,包括dagsheduler和task scheduler等组件。
(5)client:用户提交程序的入口。
23、spark工作机制?
用户在client端提交作业后,会由Driver运行main方法并创建spark context上下文。
执行add算子,形成dag图输入dagscheduler,按照add之间的依赖关系划分stage输入task scheduler。
task scheduler会将stage划分为task set分发到各个节点的executor中执行。
24、spark的优化怎么做?
通过spark-env文件、程序中sparkconf和set property设置。
(1)计算量大,形成的lineage过大应该给已经缓存了的rdd添加checkpoint,以减少容错带来的开销。
(2)小分区合并,过小的分区造成过多的切换任务开销,使用repartition。
25、kafka工作原理?
producer向broker发送事件,consumer从broker消费事件。
事件由topic区分开,每个consumer都会属于一个group。
相同group中的consumer不能重复消费事件,而同一事件将会发送给每个不同group的consumer。
26、ALS算法原理?
答:对于user-product-rating数据,als会建立一个稀疏的评分矩阵,其目的就是通过一定的规则填满这个稀疏矩阵。
als会对稀疏矩阵进行分解,分为用户-特征值,产品-特征值,一个用户对一个产品的评分可以由这两个矩阵相乘得到。
通过固定一个未知的特征值,计算另外一个特征值,然后交替反复进行最小二乘法,直至差平方和最小,即可得想要的矩阵。
27、kmeans算法原理?
随机初始化中心点范围,计算各个类别的平均值得到新的中心点。
重新计算各个点到中心值的距离划分,再次计算平均值得到新的中心点,直至各个类别数据平均值无变化。
28、canopy算法原理?
根据两个阈值来划分数据,以随机的一个数据点作为canopy中心。
计算其他数据点到其的距离,划入t1、t2中,划入t2的从数据集中删除,划入t1的其他数据点继续计算,直至数据集中无数据。
29、朴素贝叶斯分类算法原理?
对于待分类的数据和分类项,根据待分类数据的各个特征属性,出现在各个分类项中的概率判断该数据是属于哪个类别的。
30、关联规则挖掘算法apriori原理?
一个频繁项集的子集也是频繁项集,针对数据得出每个产品的支持数列表,过滤支持数小于预设值的项,对剩下的项进行全排列,重新计算支持数,再次过滤,重复至全排列结束,可得到频繁项和对应的支持数。
面试的时候这样的题目给你,你会不会说,fuck~,怎么这么变态,经常写代码的人,看到这个应该会很面熟。究竟什么样的题目这么变态,解答如下,仅供参考学习
1、Operation category READ is not supported in state standby是什么原因导致的?org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.ipc.StandbyException): Operation category READ is not supported in state standby 答:此时请登录Hadoop的管理界面查看运行节点是否处于standby
如登录地址是:http://xx.xx.xx.xx:50070/dfshealth.html#tab-overview
如果是,则不可在处于StandBy机器运行spark计算,因为该台机器为备分机器
2、不配置spark.deploy.recoveryMode选项为ZOOKEEPER,会有什么不好的地方
如果不设置spark.deploy.recoveryMode的话,那么集群的所有运行数据在Master重启是都会丢失,可参考BlackHolePersistenceEngine的实现。
3、多Master如何配置
因为涉及到多个Master,所以对于应用程序的提交就有了一点变化,因为应用程序需要知道当前的Master的IP地址和端口。这种HA方案处理这种情况很简单,只需要在SparkContext指向一个Master列表就可以了,如spark://host1:port1,host2:port2,host3:port3,应用程序会轮询列表。
4、No Space Left on the device(Shuffle临时文件过多)
由于Spark在计算的时候会将中间结果存储到/tmp目录,而目前linux又都支持tmpfs,其实就是将/tmp目录挂载到内存当中。
那么这里就存在一个问题,中间结果过多导致/tmp目录写满而出现如下错误
No Space Left on the device
解决办法
第一种:修改配置文件spark-env.sh,把临时文件引入到一个自定义的目录中去即可
export SPARK_LOCAL_DIRS=/home/utoken/datadir/spark/tmp
第二种:偷懒方式,针对tmp目录不启用tmpfs,直接修改/etc/fstab
5、java.lang.OutOfMemory, unable to create new native thread
Caused by: java.lang.OutOfMemoryError: unable to create new native thread
at java.lang.Thread.start0(Native Method)
at java.lang.Thread.start(Thread.java:640)
上面这段错误提示的本质是Linux操作系统无法创建更多进程,导致出错,并不是系统的内存不足。因此要解决这个问题需要修改Linux允许创建更多的进程,就需要修改Linux最大进程数。
[utoken@nn1 ~]ulimit−a临时修改允许打开的最大进程数[utoken@nn1 ]ulimit−a临时修改允许打开的最大进程数[utoken@nn1 ] vim /etc/security/limits.d/90-nproc.conf
soft nproc 60000
root soft nproc unlimited
永久修改用户打开文件的最大句柄数,该值默认1024,一般都会不够,常见错误就是not open file
[utoken@nn1 ~]$ vim /etc/security/limits.conf
bdata soft nofile 65536
bdata hard nofile 65536
6、Worker节点中的work目录占用许多磁盘空间
目录地址:/home/utoken/software/spark-1.3.0-bin-hadoop2.4/work
这些是Driver上传到worker的文件,需要定时做手工清理,否则会占用许多磁盘空间
7、spark-shell提交Spark Application如何解决依赖库
spark-shell的话,利用–driver-class-path选项来指定所依赖的jar文件,注意的是–driver-class-path后如果需要跟着多个jar文件的话,jar文件之间使用冒号(:)来分割。
8、Spark在发布应用的时候,出现连接不上master问题,如下
15/11/19 11:35:50 INFO AppClient$ClientEndpoint: Connecting to master spark://s1:7077…
15/11/19 11:35:50 WARN ReliableDeliverySupervisor: Association with remote system [akka.tcp://sparkMaster@s1:7077] has failed, address is now gated for [5000] ms. Reason: [Disassociated]
解决方式
检查所有机器时间是否一致、hosts是否都配置了映射、客户端和服务器端的Scala版本是否一致、Scala版本是否和Spark兼容
检查是否兼容问题请参考官方网站介绍:
9、开发spark应用程序(和Flume-NG结合时)发布应用时可能出现org.jboss.netty.channel.ChannelException: Failed to bind to: /192.168.10.156:18800
15/11/27 10:33:44 ERROR ReceiverSupervisorImpl: Stopped receiver with error: org.jboss.netty.channel.ChannelException: Failed to bind to: /192.168.10.156:18800
15/11/27 10:33:44 ERROR Executor: Exception in task 0.0 in stage 2.0 (TID 70)
org.jboss.netty.channel.ChannelException: Failed to bind to: /192.168.10.156:18800
at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:272)
Caused by: java.net.BindException: Cannot assign requested address
由于spark通过Master发布的时候,会自动选取发送到某一台的worker节点上,所以这里绑定端口的时候,需要选择相应的worker服务器,但是由于我们无法事先了解到,spark发布到哪一台服务器的,所以这里启动报错,是因为在 192.168.10.156:18800的机器上面没有启动Driver程序,而是发布到了其他服务器去启动了,所以无法监听到该机器出现问题,所以我们需要设置spark分发包时,发布到所有worker节点机器,或者发布后,我们去寻找发布到了哪一台机器,重新修改绑定IP,重新发布,有一定几率发布成功。详情可见《印象笔记-战5渣系列——Spark Streaming启动问题 - 推酷》
10、spark-shell 找不到hadoop so问题解决
[main] WARN org.apache.hadoop.util.NativeCodeLoader - Unable to load native-hadoop library for your platform… using builtin-java classes where applicable
在Spark的conf目录下,修改spark-env.sh文件,加入LD_LIBRARY_PATH环境变量,值为HADOOP的native库路径即可.
11、ERROR XSDB6: Another instance of Derby may have already booted the database /home/bdata/data/metastore_db.
在使用Hive on Spark模式操作hive里面的数据时,报以上错误,原因是因为HIVE采用了derby这个内嵌数据库作为数据库,它不支持多用户同时访问,解决办法就是把derby数据库换成mysql数据库即可
变更方式
12、java.lang.IllegalArgumentException: java.net.UnknownHostException: dfscluster
解决办法:
找不到hdfs集群名字dfscluster,这个文件在HADOOP的etc/hadoop下面,有个文件hdfs-site.xml,复制到Spark的conf下,重启即可
如:执行脚本,分发到所有的Spark集群机器中,
[bdata@bdata4 hadoop]foriin34,35,36,37,38;doscphdfs−site.xml192.168.10.i:/u01/spark-1.5.1/conf/ ; done
13、Exception in thread “main” java.lang.Exception: When running with master ‘yarn-client’ either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.
问题:在执行yarn集群或者客户端时,报以上错误,
[bdata@bdata4 bin]./spark−sql–masteryarn−clientExceptioninthread“main”java.lang.Exception:Whenrunningwithmaster‘yarn−client′eitherHADOOPCONFDIRorYARNCONFDIRmustbesetintheenvironment.解决办法根据提示,配置HADOOPCONFDIRorYARNCONFDIR的环境变量即可exportHADOOPHOME=/u01/hadoop−2.6.1exportHADOOPCONFDIR=./spark−sql–masteryarn−clientExceptioninthread“main”java.lang.Exception:Whenrunningwithmaster‘yarn−client′eitherHADOOPCONFDIRorYARNCONFDIRmustbesetintheenvironment.解决办法根据提示,配置HADOOPCONFDIRorYARNCONFDIR的环境变量即可exportHADOOPHOME=/u01/hadoop−2.6.1exportHADOOPCONFDIR=HADOOP_HOME/etc/hadoop
PATH=PATH:HOME/.local/bin:HOME/bin:SQOOP_HOME/bin:HIVEHOME/bin:HADOOP_HOME/bin
14、Job aborted due to stage failure: Task 3 in stage 0.0 failed 4 times, most recent failure: Lost task 3.3 in
[Stage 0:> (0 + 4) / 42]2016-01-15 11:28:16,512 [org.apache.spark.scheduler.TaskSchedulerImpl]-[ERROR] Lost executor 0 on 192.168.10.38: remote Rpc client disassociated
[Stage 0:> (0 + 4) / 42]2016-01-15 11:28:23,188 [org.apache.spark.scheduler.TaskSchedulerImpl]-[ERROR] Lost executor 1 on 192.168.10.38: remote Rpc client disassociated
[Stage 0:> (0 + 4) / 42]2016-01-15 11:28:29,203 [org.apache.spark.scheduler.TaskSchedulerImpl]-[ERROR] Lost executor 2 on 192.168.10.38: remote Rpc client disassociated
[Stage 0:> (0 + 4) / 42]2016-01-15 11:28:36,319 [org.apache.spark.scheduler.TaskSchedulerImpl]-[ERROR] Lost executor 3 on 192.168.10.38: remote Rpc client disassociated
2016-01-15 11:28:36,321 [org.apache.spark.scheduler.TaskSetManager]-[ERROR] Task 3 in stage 0.0 failed 4 times; aborting job
Exception in thread “main” org.apache.spark.SparkException : Job aborted due to stage failure: Task 3 in stage 0.0 failed 4 times, most recent failure: Lost task 3.3 in stage 0.0 (TID 14, 192.168.10.38): ExecutorLostFailure (executor 3 lost)
Driver stacktrace:
at org.apache.spark.scheduler.DAGScheduler.orgapacheapacheDAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1283)
解决方案
这里遇到的问题主要是因为数据源数据量过大,而机器的内存无法满足需求,导致长时间执行超时断开的情况,数据无法有效进行交互计算,因此有必要增加内存
15、长时间等待无反应,并且看到服务器上面的web界面有内存和核心数,但是没有分配,如下图
[Stage 0:> (0 + 0) / 42]
或者日志信息显示:
16/01/15 14:18:56 WARN TaskSchedulerImpl: Initial job has not accepted any resources; check your cluster UI to ensure that workers are registered and have sufficient resources
解决方案
出现上面的问题主要原因是因为我们通过参数spark.executor.memory设置的内存过大,已经超过了实际机器拥有的内存,故无法执行,需要等待机器拥有足够的内存后,才能执行任务,可以减少任务执行内存,设置小一些即可
16、内存不足或数据倾斜导致Executor Lost(spark-submit提交)
TaskSetManager: Lost task 1.0 in stage 6.0 (TID 100, 192.168.10.37): java.lang.OutOfMemoryError: Java heap space
16/01/15 14:29:51 INFO BlockManagerInfo: Added broadcast_8_piece0 in memory on 192.168.10.37:57139 (size: 42.0 KB, free: 24.2 MB)
16/01/15 14:29:53 INFO BlockManagerInfo: Added broadcast_8_piece0 in memory on 192.168.10.38:53816 (size: 42.0 KB, free: 24.2 MB)
16/01/15 14:29:55 INFO TaskSetManager: Starting task 3.0 in stage 6.0 (TID 102, 192.168.10.37, ANY, 2152 bytes)
16/01/15 14:29:55 WARN TaskSetManager: Lost task 1.0 in stage 6.0 (TID 100, 192.168.10.37): java.lang.OutOfMemoryError: Java heap space
at java.io.BufferedOutputStream.(BufferedOutputStream.java:76)
at java.io.BufferedOutputStream.(BufferedOutputStream.java:59)
…….
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 4 times, most recent failure: Lost task 0.3 in stage 6.0 (TID 142, 192.168.10.36): ExecutorLostFailure (executor 4 lost)
……
WARN TaskSetManager: Lost task 4.1 in stage 6.0 (TID 137, 192.168.10.38): java.lang.OutOfMemoryError: GC overhead limit exceeded
解决办法:
由于我们在执行Spark任务是,读取所需要的原数据,数据量太大,导致在Worker上面分配的任务执行数据时所需要的内存不够,直接导致内存溢出了,所以我们有必要增加Worker上面的内存来满足程序运行需要。
在Spark Streaming或者其他spark任务中,会遇到在Spark中常见的问题,典型如Executor Lost 相关的问题(shuffle fetch 失败,Task失败重试等)。这就意味着发生了内存不足或者数据倾斜的问题。这个目前需要考虑如下几个点以获得解决方案:
A、相同资源下,增加partition数可以减少内存问题。 原因如下:通过增加partition数,每个task要处理的数据少了,同一时间内,所有正在运行的task要处理的数量少了很多,所有Executor占用的内存也变小了。这可以缓解数据倾斜以及内存不足的压力。
B、关注shuffle read 阶段的并行数。例如reduce,group 之类的函数,其实他们都有第二个参数,并行度(partition数),只是大家一般都不设置。不过出了问题再设置一下,也不错。
C、给一个Executor 核数设置的太多,也就意味着同一时刻,在该Executor 的内存压力会更大,GC也会更频繁。我一般会控制在3个左右。然后通过提高Executor数量来保持资源的总量不变。
17、 Spark Streaming 和kafka整合后读取消息报错:OffsetOutOfRangeException
解决方案:如果和kafka消息中间件结合使用,请检查消息体是否大于默认设置1m,如果大于,则需要设置fetch.message.max.bytes=1m,这里需要把值设置大些
18、java.io.IOException : Could not locate executable null\bin\winutils.exe in the Hadoop binaries.(spark sql on hive 任务引发HiveContext NullPointerException)
解决办法
在开发hive和Spark整合的时候,如果是Windows系统,并且没有配置HADOOP_HOME的环境变量,那么可能找不到winutils.exe这个工具,由于使用hive时,对该命令有依赖,所以不要忽视该错误,否则将无法创建HiveContext,一直报Exception in thread “main” java.lang.RuntimeException: java.lang.NullPointerException
因此,解决该办法有两个方式
A、把任务打包成jar,上传到服务器上面,服务器是配置过HADOOP_HOME环境变量的,并且不需要依赖winutils,所以只需要通过spark-submit方式提交即可,如:
[bdata@bdata4 app]$ spark-submit –class com.pride.hive.HiveOnSparkTest –master spark://bdata4:7077 spark-simple-1.0.jar
B、解决winutils.exe命令不可用问题,配置Windows上面HADOOP_HOME的环境变量,或者在程序最开始的地方设置HADOOP_HOME的属性配置,这里需要注意,由于最新版本已经没有winutils这些exe命令了,我们需要在其他地方下载该命令放入HADOOP的bin目录下,当然也可以直接配置下载项目的环境变量,变量名一定要是HADOOP_HOME才行
下载地址:https://github.com/srccodes/hadoop-common-2.2.0-bin/archive/master.zip (记得翻墙哦)
任何项目都生效,需要配置Windows的环境变量,如果只在程序中生效可在程序中配置即可,如
//用于解决Windows下找不到winutils.exe命令
System. setProperty(“hadoop.home.dir”, “E:\Software\hadoop-common-2.2.0-bin” );
19、The root scratch dir: /tmp/hive on HDFS should be writable. Current permissions are: rwx——
解决办法
1、程序中设置环境变量:System.setProperty(“HADOOP_USER_NAME”, “bdata”)
2、修改HDFS的目录权限
Update the permission of your /tmp/hive HDFS directory using the following command
hadoop dfs -chmod 777 /tmp/hive
此问题暂未解决,估计是17点解决winutils有问题,建议最好把任务程序发布到服务器上面解决
20、Exception in thread “main” org.apache.hadoop.security.AccessControlException : Permission denied: user=Administrator, access=WRITE, inode=”/data”:bdata:supergroup:drwxr-xr-x
解决办法
1、在系统的环境变量或java JVM变量里面添加HADOOP_USER_NAME,如程序中添加System.setProperty(“HADOOP_USER_NAME”, “bdata”);,这里的值就是以后会运行HADOOP上的Linux的用户名,如果是eclipse,则修改完重启eclipse,不然可能不生效
2、hdfs dfs -chmod 777 修改相应权限地址
21、运行Spark-SQL报错:org.apache.spark.sql.AnalysisException: unresolved operator ‘Project
解决办法:
在Spark-sql和hive结合时或者单独Spark-sql,运行某些sql语句时,偶尔出现上面错误,那么我们可以检查一下sql的问题,这里遇到的问题是嵌套语句太多,导致spark无法解析,所以需要修改sql或者改用其他方式处理;特别注意该语句可能在hive里面没有错误,spark才会出现的一种错误。
22.在SPARKHOME/conf/spark−env.sh中设置这些变量好像也只是在terminal中的shell环境中才有效JAVAHOMEisnotsetException:Javagatewayprocessexitedbeforesendingthedriveritsportnumber但是在命令行中是有的pipi@pipicmp: SPARKHOME/conf/spark−env.sh中设置这些变量好像也只是在terminal中的shell环境中才有效JAVAHOMEisnotsetException:Javagatewayprocessexitedbeforesendingthedriveritsportnumber但是在命令行中是有的pipi@pipicmp: echo $JAVA_HOME
/home/pipi/ENV/jdk
解决方法1:在py代码中加入JAVA_HOME到os中
JAVA_HOME = /home/pipi/ENV/jdk
os.environ[‘JAVA_HOME’] = conf.get(SECTION, ‘JAVA_HOME’)
解决方法2:或者在hadoop中配置好JAVA_HOME
hadoop中配置JAVA_HOME
23.ValueError: Cannot run multiple SparkContexts at once
Welcome to
__
/ / _ _/ /__
\ \/ \/ _ `/ _/ ‘/
/_ / ._/_,// /_/_\ version 2.0.1
/_/
Using Python version 3.5.2 (default, Sep 10 2016 08:21:44)
SparkSession available as ‘spark’.
ValueError: Cannot run multiple SparkContexts at once; existing SparkContext(app=pyspark-shell, master=local) created by at