spark-submit的时候如何引入外部jar包

在通过spark-submit提交任务时，可以通过添加配置参数来指定
–driver-class-path 外部jar包 –jars 外部jar包

方法一：spark-submit –jars

根据spark官网，在提交任务的时候指定–jars，用逗号分开。这样做的缺点是每次都要指定jar包，如果jar包少的话可以这么做，但是如果多的话会很麻烦。

命令：spark-submit –master yarn-client –jars .jar,.jar

方法二：extraClassPath

提交时在spark-default中设定参数，将所有需要的jar包考到一个文件里，然后在参数中指定该目录就可以了，较上一个方便很多：

spark.executor.extraClassPath=/home/hadoop/wzq_workspace/lib/* spark.driver.extraClassPath=/home/hadoop/wzq_workspace/lib/*

需要注意的是,你要在所有可能运行spark任务的机器上保证该目录存在，并且将jar包考到所有机器上。这样做的好处是提交代码的时候不用再写一长串jar了，缺点是要把所有的jar包都拷一遍。

谈谈spark中的宽窄依赖

RDD和它依赖的父RDD（s）的关系有两种不同的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。
宽依赖：指的是多个子RDD的Partition会依赖同一个父RDD的Partition
窄依赖：指的是每一个父RDD的Partition最多被子RDD的一个Partition使用。

spark中如何划分stage

1. Spark Application中可以因为不同的Action触发众多的job，一个Application中可以有很多的job，每个job是由一个或者多个Stage构成的，后面的Stage依赖于前面的Stage，也就是说只有前面依赖的Stage计算完毕后，后面的Stage才会运行。

2. Stage划分的依据就是宽依赖，何时产生宽依赖，例如reduceByKey,groupByKey的算子，会导致宽依赖的产生。

3. 由Action（例如collect）导致了SparkContext.runJob的执行，最终导致了DAGScheduler中的submitJob的执行，其核心是通过发送一个case class JobSubmitted对象给eventProcessLoop。
   eventProcessLoop是DAGSchedulerEventProcessLoop的具体实例，而DAGSchedulerEventProcessLoop是eventLoop的子类，具体实现EventLoop的onReceive方法，onReceive方法转过来回调doOnReceive

4. 在doOnReceive中通过模式匹配的方法把执行路由到

5. 在handleJobSubmitted中首先创建finalStage，创建finalStage时候会建立父Stage的依赖链条

spark 如何防止内存溢出

driver端的内存溢出
可以增大driver的内存参数：spark.driver.memory (default 1g) 这个参数用来设置Driver的内存。
在Spark程序中，SparkContext，DAGScheduler都是运行在Driver端的。对应rdd的Stage切分也是在Driver端运行，如果用户自己写的程序有过多的步骤，切分出过多的Stage，这部分信息消耗的是Driver的内存，这个时候就需要调大Driver的内存。

map过程产生大量对象导致内存溢出
这种溢出的原因是在单个map中产生了大量的对象导致的
通过 rdd.repartition(10000).map(x=>for(i <- 1 to 10000) yield i.toString)。

shuffle后内存溢出
shuffle内存溢出的情况可以说都是shuffle后，单个文件过大导致的。
解释：
在Spark中，join，reduceByKey这一类型的过程，都会有shuffle的过程，在shuffle的使用，需要传入一个partitioner，大部分Spark中的shuffle操作，默认的partitioner都是HashPatitioner，默认值是父RDD中最大的分区数,这个参数通过spark.default.parallelism控制(在spark-sql中用spark.sql.shuffle.partitions) ， spark.default.parallelism参数只对HashPartitioner有效，所以如果是别的Partitioner或者自己实现的Partitioner就不能使用spark.default.parallelism这个参数来控制shuffle的并发量了。如果是别的partitioner导致的shuffle内存溢出，就需要从partitioner的代码增加partitions的数量。
可以通过设置Partitioner，充分区解决

standalone模式下资源分配不均匀导致内存溢出
在standalone的模式下如果配置了–total-executor-cores 和 –executor-memory 这两个参数，但是没有配置–executor-cores这个参数的话，就有可能导致，每个Executor的memory是一样的，但是cores的数量不同，那么在cores数量多的Executor中，由于能够同时执行多个Task，就容易导致内存溢出的情况。这种情况的解决方法就是同时配置–executor-cores或者spark.executor.cores参数，确保Executor资源分配均匀。

使用rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)代替rdd.cache()
rdd.cache()和rdd.persist(Storage.MEMORY_ONLY)是等价的，在内存不足的时候rdd.cache()的数据会丢失，再次使用的时候会重算，而rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)在内存不足的时候会存储在磁盘，避免重算，只是消耗点IO时间。

spark中cache和persist的区别

cache：缓存数据，默认是缓存在内存中，其本质还是调用persist persist:缓存数据，有丰富的数据缓存策略。数据可以保存在内存也可以保存在磁盘中，使用的时候指定对应的缓存级别就可以了。

Spark分布式集群搭建的步骤

spark中的数据倾斜的现象、原因、后果

(1)、数据倾斜的现象
多数task执行速度较快,少数task执行时间非常长，或者等待很长时间后提示你内存不足，执行失败。
(2)、数据倾斜的原因
数据问题
1、key本身分布不均衡（包括大量的key为空） 2、key的设置不合理 spark使用问题
1、shuffle时的并发度不够 2、计算方式有误

(3)、数据倾斜的后果
1、spark中的stage的执行时间受限于最后那个执行完成的task,因此运行缓慢的任务会拖垮整个程序的运行速度（分布式程序运行的速度是由最慢的那个task决定的）。 2、过多的数据在同一个task中运行，将会把executor撑爆。

如何解决spark中的数据倾斜问题

发现数据倾斜的时候，不要急于提高executor的资源，修改参数或是修改程序，首先要检查数据本身，是否存在异常数据。

1、数据问题造成的数据倾斜
找出异常的key
如果任务长时间卡在最后最后1个(几个)任务，首先要对key进行抽样分析，判断是哪些key造成的。
选取key，对数据进行抽样，统计出现的次数，根据出现次数大小排序取出前几个。 比如: df.select(“key”).sample(false,0.1).(k=>(k,1)).reduceBykey(+).map(k=>(k._2,k._1)).sortByKey(false).take(10) 如果发现多数数据分布都较为平均，而个别数据比其他数据大上若干个数量级，则说明发生了数据倾斜。
经过分析，倾斜的数据主要有以下三种情况:

1、null（空值）或是一些无意义的信息()之类的,大多是这个原因引起。
2、无效数据，大量重复的测试数据或是对结果影响不大的有效数据。
3、有效数据，业务导致的正常数据分布。

解决办法
第1，2种情况，直接对数据进行过滤即可（因为该数据对当前业务不会产生影响）。 第3种情况则需要进行一些特殊操作，常见的有以下几种做法

(1) 隔离执行，将异常的key过滤出来单独处理，最后与正常数据的处理结果进行union操作。
(2) 对key先添加随机值，进行操作后，去掉随机值，再进行一次操作。
(3) 使用reduceByKey 代替 groupByKey(reduceByKey用于对每个key对应的多个value进行merge操作，最重要的是它能够在本地先进行merge操作，并且merge操作可以通过函数自定义.)
(4) 使用map join。
案例
如果使用reduceByKey因为数据倾斜造成运行失败的问题。具体操作流程如下:
(1) 将原始的 key 转化为 key + 随机值(例如Random.nextInt) (2) 对数据进行 reduceByKey(func) (3) 将 key + 随机值 转成 key (4) 再对数据进行 reduceByKey(func)

2、spark使用不当造成的数据倾斜
提高shuffle并行度

dataFrame和sparkSql可以设置spark.sql.shuffle.partitions参数控制shuffle的并发度，默认为200。 rdd操作可以设置spark.default.parallelism控制并发度，默认参数由不同的Cluster Manager控制。 局限性: 只是让每个task执行更少的不同的key。无法解决个别key特别大的情况造成的倾斜，如果某些key的大小非常大，即使一个task单独执行它，也会受到数据倾斜的困扰。

使用map join 代替reduce join

在小表不是特别大(取决于你的executor大小)的情况下使用，可以使程序避免shuffle的过程，自然也就没有数据倾斜的困扰了. 局限性: 因为是先将小数据发送到每个executor上，所以数据量不能太大。

flume整合sparkStreaming问题

如何实现sparkStreaming读取flume中的数据
有两种方式push和pull方式：

push方式

  LogLevel.setStreamingLogLevels()
    val Array(host, port) = args
    val batchInterval = Milliseconds(2000)
    // Create the context and set the batch size
    val sparkConf = new SparkConf().setAppName("FlumeEventCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, batchInterval)
    // Create a flume stream
    val stream = FlumeUtils.createStream(ssc, host, port.toInt, StorageLevel.MEMORY_ONLY_SER_2)
    // Print out the count of events received from this server in each batch
    stream.count().map(cnt => "Received " + cnt + " flume events.").print()
    //拿到消息中的event,从event中拿出body,body是真正的消息体
    stream.flatMap(t=>{new String(t.event.getBody.array()).split(" ")}).map((_,1)).reduceByKey(_+_).print

    ssc.start()
    ssc.awaitTermination()

pull 方式

 //当sink有多个的时候
    val flumesinklist = Array[InetSocketAddress](new InetSocketAddress("mini1", 8888))
    val flumeStream = FlumeUtils.createPollingStream(ssc, flumesinklist, StorageLevel.MEMORY_ONLY_2)

    flumeStream.count().map(cnt => "Received " + cnt + " flume events.").print()
    flumeStream.flatMap(t => {
      new String(t.event.getBody.array()).split(" ")
    }).map((_, 1)).reduceByKey(_ + _).print()


    // Print out the count of events received from this server in each batch
    //stream.count().map(cnt => "Received " + cnt + " flume events.").print()
    //拿到消息中的event,从event中拿出body,body是真正的消息体
    //stream.flatMap(t=>{new String(t.event.getBody.array()).split(" ")}).map((_,1)).reduceByKey(_+_).print

    ssc.start()
    ssc.awaitTermination()

如何保证flume数据不丢失

1.flume那边采用的channel是将数据落地到磁盘中，保证数据源端安全性（可以在补充一下，flume在这里的channel可以设置为memory内存中，提高数据接收处理的效率，但是由于数据在内存中，安全机制保证不了，故选择channel为磁盘存储。整个流程运行有一点的延迟性）

1. sparkStreaming通过拉模式整合的时候，使用了FlumeUtils这样一个类，该类是需要依赖一个额外的jar包（spark-streaming-flume_2.10）
   3.利用StreamingContext.getOrCreate（checkpoint, creatingFunc: () => StreamingContext）来创建一个StreamingContext，设置checkpoin地址

4.spark.streaming.receiver.writeAheadLogs.enable这个property设置为true。

Spark Streaming的数据可靠性

有了checkpoint机制、write ahead log机制、Receiver缓存机器、可靠的Receiver（即数据接收并备份成功后会发送ack），可以保证无论是worker失效还是driver失效，都是数据0丢失。原因是：如果没有Receiver服务的worker失效了，RDD数据可以依赖血统来重新计算；如果Receiver所在worker失败了，由于Reciever是可靠的，并有write ahead log机制，则收到的数据可以保证不丢；如果driver失败了，可以从checkpoint中恢复数据重新构建。

kafka整合sparkStreaming问题

receiver 方式
val kafkaStream = KafkaUtils.createStream(streamingContext,
[ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

注意点：
1.在Receiver的方式中，Spark中的partition和kafka中的partition并不是相关的，
所以如果我们加大每个topic的partition数量，仅仅是增加线程来处理由单一Receiver消费的主题。
但是这并没有增加Spark在处理数据上的并行度.
对于不同的Group和topic我们可以使用多个Receiver创建不同的Dstream来并行接收数据，
之后可以利用union来统一成一个Dstream

2.可能因为execture ，driver死掉而丢失
3.增加(spark.streaming.receiver.writeAheadLog.enable=true）
KafkaUtils.createStream(…, StorageLevel.MEMORY_AND_DISK_SER)
减少数据丢失

Direct 方式
//创建DStream，返回接收到的输入数据
var stream=KafkaUtils.createDirectStream[String,String](ssc, PreferConsistent,SubscribeString,String)

两种方式优缺点：
receiver 保证数据不丢失，但是可能数据会重复消费
Direct 相比Receiver模式而言能够确保机制更加健壮. 区别于使用Receiver来被动接收数据, Direct模式会周期性地主动查询Kafka, 来获得每个topic+partition的最新的offset, 从而定义每个batch的offset的范围. 当处理数据的job启动时, 就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

receiver 对于多个分区需要增加多个recevier然后union使用
Direct 不需要创建多个输入DStream然后对它们进行union操作. Spark会创建跟Kafka partition一样多的RDD partition, 并且会并行从Kafka中读取数据. 所以在Kafka partition和RDD partition之间, 有一个一对一的映射关系

receiver没有direc高性能，dircet不需要设置spark.streaming.receiver.writeAheadLog.enable=true，而是利用kafka本身的数据备份保证数据的唯一性，

Spark master使用zookeeper进行HA的，有哪些元数据保存在Zookeeper？

答：spark通过这个参数spark.deploy.zookeeper.dir指定master元数据在zookeeper中保存的位置，包括Worker，Driver和Application以及Executors。standby节点要从zk中，获得元数据信息，恢复集群运行状态，才能对外继续提供服务，作业提交资源申请等，在恢复前是不能接受请求的。另外，Master切换需要注意2点

1）在Master切换的过程中，所有的已经在运行的程序皆正常运行！因为Spark Application在运行前就已经通过Cluster Manager获得了计算资源，所以在运行时Job本身的调度和处理和Master是没有任何关系的！

2） 在Master的切换过程中唯一的影响是不能提交新的Job：一方面不能够提交新的应用程序给集群，因为只有Active Master才能接受新的程序的提交请求；另外一方面，已经运行的程序中也不能够因为Action操作触发新的Job的提交请求；

Spark master HA 主从切换过程不会影响集群已有的作业运行，为什么？

因为程序在运行之前，已经申请过资源了，driver和Executors通讯，不需要和master进行通讯的。

如何配置spark master的HA？

1)配置zookeeper

2)修改spark_env.sh文件,spark的master参数不在指定，添加如下代码到各个master节点

export SPARK_DAEMON_JAVA_OPTS=”-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=zk01:2181,zk02:2181,zk03:2181 -Dspark.deploy.zookeeper.dir=/spark”

1. 将spark_env.sh分发到各个节点

4)找到一个master节点，执行./start-all.sh，会在这里启动主master,其他的master备节点，启动master命令: ./sbin/start-master.sh

5)提交程序的时候指定master的时候要指定三台master，例如

./spark-shell –master spark://master01:7077,master02:7077,master03:7077

driver的功能是什么？

1）一个Spark作业运行时包括一个Driver进程，也是作业的主进程，具有main函数，并且有SparkContext的实例，是程序的人口点；2）功能：负责向集群申请资源，向master注册信息，负责了作业的调度，，负责作业的解析、生成Stage并调度Task到Executor上。包括DAGScheduler，TaskScheduler。

Spark中Work的主要工作是什么？

答：主要功能：管理当前节点内存，CPU的使用状况，接收master分配过来的资源指令，通过ExecutorRunner启动程序分配任务，worker就类似于包工头，管理分配新进程，做计算的服务，相当于process服务。需要注意的是：1）worker会不会汇报当前信息给master，worker心跳给master主要只有workid，它不会发送资源信息以心跳的方式给mater，master分配的时候就知道work，只有出现故障的时候才会发送资源。2）worker不会运行代码，具体运行的是Executor是可以运行具体appliaction写的业务逻辑代码，操作代码的节点，它不会运行程序的代码的。

Spark为什么比mapreduce快？

1）基于内存计算，减少低效的磁盘交互；
2）高效的调度算法，基于DAG；
3)容错机制Linage，精华部分就是DAG和Lingae

Mapreduce和Spark的都是并行计算，那么他们有什么相同和区别

答：两者都是用mr模型来进行并行计算:

1)hadoop的一个作业称为job，job里面分为map task和reduce task，每个task都是在自己的进程中运行的，当task结束时，进程也会结束。

2)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可以快速启动读取内存进行计算。

3)hadoop的job只有map和reduce操作，表达能力比较欠缺而且在mr过程中会重复的读写hdfs，造成大量的io操作，多个job需要自己管理关系。

spark的迭代计算都是在内存中进行的，API中提供了大量的RDD操作如join，groupby等，而且通过DAG图可以实现良好的容错。

spark工作机制？

用户在client端提交作业后，会由Driver运行main方法并创建spark context上下文。

执行add算子，形成dag图输入dagscheduler，按照add之间的依赖关系划分stage输入task scheduler。 task scheduler会将stage划分为task set分发到各个节点的executor中执行。

cache后面能不能接其他算子,它是不是action操作？

cache可以接其他算子，但是接了算子之后，起不到缓存应有的效果，因为会重新触发cache。

cache不是action操作

RDD的弹性表现在哪几点？

1）自动的进行内存和磁盘的存储切换；

2）基于Lingage的高效容错；

3）task如果失败会自动进行特定次数的重试；

4）stage如果失败会自动进行特定次数的重试，而且只会计算失败的分片；

5）checkpoint和persist，数据计算之后持久化缓存

6）数据调度弹性，DAG TASK调度和资源无关

7）数据分片的高度弹性，a.分片很多碎片可以合并成大的，b.par

RDD通过Linage（记录数据更新）的方式为何很高效？

1）lazy记录了数据的来源，RDD是不可变的，且是lazy级别的，且rDD

之间构成了链条，lazy是弹性的基石。由于RDD不可变，所以每次操作就

产生新的rdd，不存在全局修改的问题，控制难度下降，所有有计算链条

将复杂计算链条存储下来，计算的时候从后往前回溯

900步是上一个stage的结束，要么就checkpoint

2）记录原数据，是每次修改都记录，代价很大

如果修改一个集合，代价就很小，官方说rdd是

粗粒度的操作，是为了效率，为了简化，每次都是

操作数据集合，写或者修改操作，都是基于集合的

rdd的写操作是粗粒度的，rdd的读操作既可以是粗粒度的

也可以是细粒度，读可以读其中的一条条的记录。

3）简化复杂度，是高效率的一方面，写的粗粒度限制了使用场景

如网络爬虫，现实世界中，大多数写是粗粒度的场景

Spark的Shuffle过程介绍

http://www.cnblogs.com/jxhd1/p/6528540.html

collect功能是什么，其底层是怎么实现的？

driver通过collect把集群中各个节点的内容收集过来汇总成结果，collect返回结果是Array类型的，collect把各个节点上的数据抓过来，抓过来数据是Array型，collect对Array抓过来的结果进行合并，合并后Array中只有一个元素，是tuple类型（KV类型的）的。

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程序设置生效

为什么Spark Application在没有获得足够的资源，job就开始执行了，可能会导致什么什么问题发生?

会导致执行该job时候集群资源不足，导致执行job结束也没有分配足够的资源，分配了部分Executor，该job就开始执行task，应该是task的调度线程和Executor资源申请是异步的；如果想等待申请完所有的资源再执行job的：需要将spark.scheduler.maxRegisteredResourcesWaitingTime设置的很大；spark.scheduler.minRegisteredResourcesRatio 设置为1，但是应该结合实际考虑

否则很容易出现长时间分配不到资源，job一直不能运行的情况。

.map与flatMap的区别

map：对RDD每个元素转换，文件中的每一行数据返回一个数组对象

flatMap：对RDD每个元素转换，然后再扁平化

将所有的对象合并为一个对象，文件中的所有行数据仅返回一个数组

对象，会抛弃值为null的值

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，框架默认将数据持久化到磁盘，这个是框架自动做的。

为什么要进行序列化

序列化可以减少数据的体积，减少存储空间，高效存储和传输数据，不好的是使用的时候要反序列化，非常消耗CPU

介绍一下join操作优化经验？

join其实常见的就分为两类： map-side join 和 reduce-side join。当大表和小表join时，用map-side join能显著提高效率。将多份数据进行关联是数据处理过程中非常普遍的用法，不过在分布式计算系统中，这个问题往往会变的非常麻烦，因为框架提供的 join 操作一般会将所有数据根据 key 发送到所有的 reduce 分区中去，也就是 shuffle 的过程。造成大量的网络以及磁盘IO消耗，运行效率极其低下，这个过程一般被称为 reduce-side-join。如果其中有张表较小的话，我们则可以自己实现在 map 端实现数据关联，跳过大量数据进行 shuffle 的过程，运行时间得到大量缩短，根据不同数据可能会有几倍到数十倍的性能提升。

介绍一下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,)，

场景：表关联查询

Spark使用parquet文件存储格式能带来哪些好处

1. 如果说HDFS 是大数据时代分布式文件系统首选标准，那么parquet则是整个大数据时代文件存储格式实时首选标准

2. 速度更快：从使用spark sql操作普通文件CSV和parquet文件速度对比上看，绝大多数情况

会比使用csv等普通文件速度提升10倍左右，在一些普通文件系统无法在spark上成功运行的情况

下，使用parquet很多时候可以成功运行

1. parquet的压缩技术非常稳定出色，在spark sql中对压缩技术的处理可能无法正常的完成工作

（例如会导致lost task，lost executor）但是此时如果使用parquet就可以正常的完成

1. 极大的减少磁盘I/o,通常情况下能够减少75%的存储空间，由此可以极大的减少spark sql处理

数据的时候的数据输入内容，尤其是在spark1.6x中有个下推过滤器在一些情况下可以极大的

减少磁盘的IO和内存的占用，（下推过滤器）

1. spark 1.6x parquet方式极大的提升了扫描的吞吐量，极大提高了数据的查找速度spark1.6和spark1.5x相比而言，提升了大约1倍的速度，在spark1.6X中，操作parquet时候cpu也进行了极大的优化，有效的降低了cpu

2. 采用parquet可以极大的优化spark的调度和执行。我们测试spark如果用parquet可以有效的减少stage的执行消耗，同时可以优化执行路径

Executor之间如何共享数据？

基于hdfs或者基于tachyon

Spark累加器有哪些特点？

1）累加器在全局唯一的，只增不减，记录全局集群的唯一状态

2）在exe中修改它，在driver读取

3）executor级别共享的，广播变量是task级别的共享

两个application不可以共享累加器，但是同一个app不同的job可以共享

spark hashParitioner的弊端是什么？

HashPartitioner分区的原理很简单，对于给定的key，计算其hashCode，并除于分区的个数取余，如果余数小于0，则用余数+分区的个数，最后返回的值就是这个key所属的分区ID；弊端是数据不均匀，容易导致数据倾斜，极端情况下某几个分区会拥有rdd的所有数据

RangePartitioner分区的原理?

RangePartitioner分区则尽量保证每个分区中数据量的均匀，而且分区与分区之间是有序的，也就是说一个分区中的元素肯定都是比另一个分区内的元素小或者大；但是分区内的元素是不能保证顺序的。简单的说就是将一定范围内的数映射到某一个分区内。其原理是水塘抽样。可以参考这篇博文

https://www.iteblog.com/archives/1522.html

介绍parition和block有什么关联关系？

1）hdfs中的block是分布式存储的最小单元，等分，可设置冗余，这样设计有一部分磁盘空间的浪费，但是整齐的block大小，便于快速找到、读取对应的内容；
2）Spark中的partion是弹性分布式数据集RDD的最小单元，RDD是由分布在各个节点上的partion组成的。partion是指的spark在计算过程中，生成的数据在计算空间内最小单元，同一份数据（RDD）的partion大小不一，数量不定，是根据application里的算子和最初读入的数据分块数量决定；
3）block位于存储空间、partion位于计算空间，block的大小是固定的、partion大小是不固定的，是从2个不同的角度去看数据。

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上运行，运行完毕释放所有资源。

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

spark中task有几种类型？

1）result task类型，最后一个task，
2）是shuffleMapTask类型，除了最后一个task都是

什么是二次排序，你是如何用spark实现二次排序的？（互联网公司常面）

http://blog.csdn.net/sundujing/article/details/51399606

class SecondarySort(val first:Int, val second:Int) extends Ordered[SecondarySort] with Serializable{
override def compare(that: SecondarySort): Int = {
  if(this.first - that.first != 0)
  {
    this.first - that.first
  } else {
    this.second - that.second
  }

`}
}


object SecondarySortApp {
def main (args: Array[String]) {
//第一步；创建spark的配置对象sparkconf

val conf = new SparkConf()//创建sparkconf对象
conf.setAppName("SecondarySortApp")//设置应用程序的名称
conf.setMaster("local")//设置本地运行

//创建sparkcontext对象，sparkcontext是程序的唯一入口

val sc = new SparkContext(conf)

val lines = sc.textFile("D:\\JavaWorkspaces\\sparkproject\\sparktest.txt")

val pairWithSortkey = lines.map(line =>(
  new SecondarySort( line.split(" ")(0).toInt,line.split(" ")(1).toInt),line
))

  val sorted = pairWithSortkey.sortByKey(false)

    val sortedResult = sorted.map(sortedline => sortedline._2)
      sortedResult.collect.foreach(println)
    }
}

如何使用Spark解决TopN问题？（互联网公司常面）

函数partitionFunc、keyFunc准备好之后，我们可以开始调用repartitionAndSortWithinPartitions：
umPartitions值为10，该值取决于分区（产品线）的个数；ascending值为False，该值表示分区内排序时使用降序。

如何使用Spark解决分组排序问题？（互联网公司常面）

1、对上述数据按key值进行分组

2、对分组后的值进行排序

3、截取分组后值得top 3位以key-value形式返回结果

窄依赖父RDD的partition和子RDD的parition是不是都是一对一的关系？

不一定，除了一对一的窄依赖，还包含一对固定个数的窄依赖（就是对父RDD的依赖的Partition的数量不会随着RDD数量规模的改变而改变），比如join操作的每个partiion仅仅和已知的partition进行join，这个join操作是窄依赖，依赖固定数量的父rdd，因为是确定的partition关系

不需要排序的hash shuffle是否一定比需要排序的sort shuffle速度快？

不一定！！当数据规模小，Hash shuffle快于Sorted Shuffle数据规模大的时候；当数据量大，sorted Shuffle会比Hash shuffle快很多，因为数量大的有很多小文件，不均匀，甚至出现数据倾斜，消耗内存大，1.x之前spark使用hash，适合处理中小规模，1.x之后，增加了Sorted shuffle，Spark更能胜任大规模处理了。

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会产生10100=1000个文件，conslidate产生1010=100个文件

备注：conslidate部分减少了文件和文件句柄，并行读很高的情况下（task很多时）还是会很多文件

Sort-basesd shuffle产生多少个临时文件

2*Map阶段所有的task数量，Mapper阶段中并行的Partition的总数量，其实就是Mapper端task
https://blog.csdn.net/kxr0502/article/details/50616805

Sort-based shuffle的缺陷?

1. 如果mapper中task的数量过大，依旧会产生很多小文件，此时在shuffle传递数据的过程中reducer段，reduce会需要同时大量的记录进行反序列化，导致大量的内存消耗和GC的巨大负担，造成系统缓慢甚至崩溃

2）如果需要在分片内也进行排序，此时需要进行mapper段和reducer段的两次排序

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更小一点。

介绍一下你对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).

描述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的注册，开始进行任务的调度和计算，直到
任务完成。

image

Yarn中的container是由谁负责销毁的，在Hadoop Mapreduce中container可以复用么？

ApplicationMaster负责销毁，在Hadoop Mapreduce不可以复用，在spark on yarn程序container可以复用

提交任务时，如何指定Spark Application的运行模式？

1）cluster模式：./spark-submit –class xx.xx.xx –master yarn –deploy-mode cluster xx.jar

1. client模式:./spark-submit –class xx.xx.xx –master yarn –deploy-mode client xx.jar

不启动Spark集群Master和work服务，可不可以运行Spark程序？

可以，只要资源管理器第三方管理就可以，如由yarn管理，spark集群不启动也可以使用spark；spark集群启动的是work和master，这个其实就是资源管理框架，yarn中的resourceManager相当于master，NodeManager相当于worker，做计算是Executor，和spark集群的work和manager可以没关系，归根接底还是JVM的运行，只要所在的JVM上安装了spark就可以。

Spark中的4040端口由什么功能?

收集Spark作业运行的信息

spark on yarn Cluster 模式下，ApplicationMaster和driver是在同一个进程么？

是,driver 位于ApplicationMaster进程中。该进程负责申请资源，还负责监控程序、资源的动态情况。

Spark on Yarn 模式有哪些优点？

1)与其他计算框架共享集群资源（eg.Spark框架与MapReduce框架同时运行，如果不用Yarn进行资源分配，MapReduce分到的内存资源会很少，效率低下）；资源按需分配，进而提高集群资源利用等。

2)相较于Spark自带的Standalone模式，Yarn的资源分配更加细致

3)Application部署简化，例如Spark，Storm等多种框架的应用由客户端提交后，由Yarn负责资源的管理和调度，利用Container作为资源隔离的单位，以它为单位去使用内存,cpu等。

4)Yarn通过队列的方式，管理同时运行在Yarn集群中的多个服务，可根据不同类型的应用程序负载情况，调整对应的资源使用量，实现资源弹性管理。

谈谈你对container的理解？

1）Container作为资源分配和调度的基本单位，其中封装了的资源如内存，CPU，磁盘，网络带宽等。 目前yarn仅仅封装内存和CPU

2)Container由ApplicationMaster向ResourceManager申请的，由ResouceManager中的资源调度器异步分配给ApplicationMaster

1. Container的运行是由ApplicationMaster向资源所在的NodeManager发起的，Container运行时需提供内部执行的任务命令.

运行在yarn中Application有几种类型的container？

1） 运行ApplicationMaster的Container：这是由ResourceManager（向内部的资源调度器）申请和启动的，用户提交应用程序时，可指定唯一的ApplicationMaster所需的资源；

2） 运行各类任务的Container：这是由ApplicationMaster向ResourceManager申请的，并由ApplicationMaster与NodeManager通信以启动之。

Spark on Yarn架构是怎么样的？（要会画哦，这个图）

image

Executor启动时，资源通过哪几个参数指定？

1)num-executors是executor的数量

2)executor-memory 是每个executor使用的内存

3)executor-cores 是每个executor分配的CPU

Mapreduce的执行过程?

阶段1：input/map/partition/sort/spill

阶段2：mapper端merge

阶段3：reducer端merge/reduce/output
详细过程参考这个http://www.cnblogs.com/hipercomer/p/4516581.html

image

一个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决定。

你的项目提交到job的时候数据量有多大？

1）回答出数据是什么格式，有没有采用什么压缩，采用了压缩的话，压缩比大概是多少；2）文件大概多大：大概起了多少个map，起了多少个reduce，map阶段读取了多少数据，reduce阶段读取了多少数据，程序大约执行了多久，3）集群什么规模，集群有多少节点，多少内存，多少CPU核数等。把这些点回答进去，而不是给个数字了事。

如何杀死一个正在运行的job

杀死一个job

MRV1：Hadoop job kill jobid

YARN: yarn application -kill applicationId

列出你所知道的调度器，说明其工作原理

a) Fifo schedular 默认的调度器 先进先出

b) Capacity schedular 计算能力调度器 选择占用内存小 优先级高的

c) Fair schedular 调肚脐 公平调度器 所有job 占用相同资源

导致Executor产生FULL gc 的原因，可能导致什么问题？

可能导致Executor僵死问题，海量数据的shuffle和数据倾斜等都可能导致full gc。以shuffle为例，伴随着大量的Shuffle写操作，JVM的新生代不断GC，Eden Space写满了就往Survivor Space写，同时超过一定大小的数据会直接写到老生代，当新生代写满了之后，也会把老的数据搞到老生代，如果老生代空间不足了，就触发FULL GC，还是空间不够，那就OOM错误了，此时线程被Blocked，导致整个Executor处理数据的进程被卡住

Spark执行任务时出现java.lang.OutOfMemoryError: GC overhead limit exceeded和java.lang.OutOfMemoryError: java heap space原因和解决方法？

原因：加载了太多资源到内存，本地的性能也不好，gc时间消耗的较多

解决方法：

1）增加参数，-XX:-UseGCOverheadLimit，关闭这个特性，同时增加heap大小，-Xmx1024m

2）下面这个两个参数调大点

export SPARK_EXECUTOR_MEMORY=6000M

export SPARK_DRIVER_MEMORY=7000M

可以参考这个：http://www.cnblogs.com/hucn/p/3572384.html

hadoop的TextInputFormat作用是什么，如何自定义实现

InputFormat会在map操作之前对数据进行两方面的预处理

1是getSplits，返回的是InputSplit数组，对数据进行split分片，每片交给map操作一次

2是getRecordReader，返回的是RecordReader对象，对每个split分片进行转换为key-value键值对格式传递给map

常用的InputFormat是TextInputFormat，使用的是LineRecordReader对每个分片进行键值对的转换，以行偏移量作为键，行内容作为值

自定义类继承InputFormat接口，重写createRecordReader和isSplitable方法

在createRecordReader中可以自定义分隔符

Hbase行健列族的概念，物理模型，表的设计原则？

行健：是hbase表自带的，每个行健对应一条数据。

列族：是创建表时指定的，为列的集合，每个列族作为一个文件单独存储，存储的数据都是字节数组，其中的数据可以有很多，通过时间戳来区分。

物理模型：整个hbase表会拆分为多个region，每个region记录着行健的起始点保存在不同的节点上，查询时就是对各个节点的并行查询，当region很大时使用.META表存储各个region的起始点，-ROOT又可以存储.META的起始点。

rowkey的设计原则：各个列簇数据平衡，长度原则、相邻原则，创建表的时候设置表放入regionserver缓存中，避免自动增长和时间，使用字节数组代替string，最大长度64kb，最好16字节以内，按天分表，两个字节散列，四个字节存储时分毫秒。

列族的设计原则：尽可能少（按照列族进行存储，按照region进行读取，不必要的io操作），经常和不经常使用的两类数据放入不同列族中，列族名字尽可能短。

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对象

spark的优化怎么做？

通过spark-env文件、程序中sparkconf和set property设置。

（1）计算量大，形成的lineage过大应该给已经缓存了的rdd添加checkpoint，以减少容错带来的开销。

（2）小分区合并，过小的分区造成过多的切换任务开销，使用repartition。

ALS算法原理？

对于user-product-rating数据，als会建立一个稀疏的评分矩阵，其目的就是通过一定的规则填满这个稀疏矩阵。

als会对稀疏矩阵进行分解，分为用户-特征值，产品-特征值，一个用户对一个产品的评分可以由这两个矩阵相乘得到。

通过固定一个未知的特征值，计算另外一个特征值，然后交替反复进行最小二乘法，直至差平方和最小，即可得想要的矩阵。