Spark面试题（2）

面试题1：Spark中谈谈你对RDD的理解？

（1）RDD：Resilient Distributed Datasets（弹性分布式数据集），首先说说RDD的作用，它的出现大大降低了开发分布式应用程序的门槛，比如在其他框架开发分布式应用程序，你需要考虑数据的拆分，数据的隔离，节点之间的通信机制，job的调度，序列化等等。其次，RDD他是一个基础的抽象类，是不可变的，可拆分的，可并行操作的数据集合。

（2）RDD的五大特性：

1）分区列表（a list of partitions）

详解：Spark中的RDD是被分区的，每一个分区都会被计算任务（task数量=partition数量）处理，所以分区数决定了Spark的并行度，RDD的并行度默认是从父RDD穿个子RDD的。默认情况下，HDFS上一个block就是一个partition，RDD的分片数量决定了并行计算的力度，可以在创建RDD时指定RDD的分片个数（textFile（“xxx”），3），如果不指定分区数量，当RDD从集合创建时，则默认分区数量为该程序所分配到的core（一个core可以承载2-4个分区），HDFS上的文件读取出来有多少个block就有多少个partition数量。

2）每个分区都有一个计算函数（ a function for computing each split）

详解：每个分区都会有计算函数，Spark的RDD的计算函数是以分片（split）为基本单位的，每个RDD都会实现compute函数，对具体的分片进行计算，因为RDD中的分片是并行的。所以计算也是并行的，这也解释了分布式计算的特性，由于RDD具有前后依赖关系，遇到宽依赖关系，如reduceBykey等操作时候，就会划分成stage，stage内部的操作都是通过Pipeline进行的，在具体处理数据时它会通过Blockmanager来获取相关的数据，因为具体的split要从外界读数据，也要把具体的计算结果写入外界，所以用了一个管理器，具体的split都会映射成BlockManager的Block，并且split会被函数处理，函数处理的具体形式是以任务的形式进行的。

3）依赖与其他的RDD列表（a list of dependencies on other RDDS）

详解：RDD会保存它的依赖关系，这有助于提高的RDD的容错性，当一个中间的分区丢失了，它能根据依赖关系快速的恢复。

这里就出来了窄依赖（父RDD的partitions最多被子RDD使用一次）和宽依赖（父RDD的partitions最少被子RDD使用一次）的概念，宽依赖就意味着有shuffle操作，Spark中的宽依赖支持两种 Shuffle Manager，即 Hash Shuffemanager和Sortshuffemanager，前者是基于Hash的 Shuffle机制，后者是基于排序的 Shuffle机制。

4）可选地，当数据类型为key-value类型时（a Partitioner for key- alue RDDS）控制分区策略和分区数

详解：每个key- alue形式的RDD都有 Partitioner属性，它决定了RDD如何分区。RDD的分片函数可以分区( Partitioner)，可传入相关的参数，如 Hash Partitioner和 Range Partitioner，它本身针对key- value的形式，如果不是key-ale的形式它就不会有具有Partitioner， Partitioner本身决定了下一步会产生多少并行的分片，同时它本身也决定了当前并行( Parallelize) Shuffle输出的并行数据，从而使Spak具有能够控制数据在不同结点上分区的特性，用户可以自定义分区策略，如Hash分区等。 spark提供了 partition By运算符，能通过集群对RDD进行数据再分配来创建一个新的RDD。

5）可选地，每个分区都有一个优先位置列表（a list of preferred locations to compute each split on）、

详解：优先位置列表会存储每个Partition的优先位置，对于一个HDFS文件来说，就是每个Partition块的位置，在Spark运行的控制台就会发现，Spark进行计算和分片之前，它就知道task发生在那个节点上，也就是说任务本身是计算层面的、代码层面的，代码在发生运算之前就已经知道它要运算的数据在什么地方了，这体现了老大的一句：数据尽量不要动，代码动的原则，Spark本身在进行任务调度时会尽可能地将任务分配到处理数据的数据块所在的具体位置。

面试题2：Spark中spark-submit的时候如何引入外部jar包？

（1）./spark-submit --jars xxxx

面试题3：spark 如何防止内存溢出（OOM）？

总的来说Spark中的内存溢出不外乎俩种情况：

（1）map执行中OOM

（2）shuffle后内存溢出

https://blog.csdn.net/yhb315279058/article/details/51035631 好好看看

driver端的内存溢出

可以增大driver的内存参数：spark.driver.memory (default 1g)

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

map过程产生大量对象导致内存溢出

这种溢出的原因是在单个map中产生了大量的对象导致的，例如：rdd.map(x=>for(i <- 1 to 10000) yield i.toString)，这个操作在rdd中，每个对象都产生了10000个对象，这肯定很容易产生内存溢出的问题。针对这种问题，在不增加内存的情况下，可以通过减少每个Task的大小，以便达到每个Task即使产生大量的对象Executor的内存也能够装得下。具体做法可以在会产生大量对象的map操作之前调用repartition方法，分区成更小的块传入map。例如：rdd.repartition(10000).map(x=>for(i <- 1 to 10000) yield i.toString)。面对这种问题注意，不能使用rdd.coalesce方法，这个方法只能减少分区，不能增加分区，不会有shuffle的过程。

数据不平衡导致内存溢出

数据不平衡除了有可能导致内存溢出外，也有可能导致性能的问题，解决方法和上面说的类似，就是调用repartition重新分区。这里就不再累赘了。

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的数量。

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时间。

2）在（url,user）的键值对中，如
a.text
127.0.0.1 xiaozhang
127.0.0.1 xiaoli
127.0.0.2 wangwu
127.0.0.3 lisi
…..
B.text
127.0.0.4 lixiaolu
127.0.0.5 lisi
127.0.0.3 zhangsan
每个文件至少有1000万行，请用程序完成一下工作，
1.各个文件的ip数

(1)1000万行数据会不会出现driverOOM呢?（空间）

(2)如何去除脏数据呢?（可靠）

2.出现在b.text而没有出现在a.text的IP

3.每个user出现的次数以及每个user对应的IP的个数

4.对应IP数最多的前K个user

package HomeWork.csdn_InQu2import org.apache.spark.sql.{Row, SparkSession, types}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}object LogAnalyDF {/*val sparkconf=new SparkConf().setMaster("local[2]").setAppName("LogAnaly")val  sc=new SparkContext(sparkconf)val ardd=sc.textFile("E:\\若泽数据\\零基础大数据篇第三期\\Hadoop综合编程\\a.txt")val lines=ardd.filter(_!="").flatMap(_.split("\t")).map(x=>Row(x(0))).foreach(println)
*/def main(args: Array[String]): Unit = {val spark = SparkSession.builder().master("local[2]").appName("RDDtoDF1").getOrCreate()val texta = spark.sparkContext.textFile("E:\\若泽数据\\零基础大数据篇第三期\\Hadoop综合编程\\a.txt").filter(_ != "").map(_.split(" ")).map(x => Row(x(0).trim, x(1).trim))//这个Row就是将度进来的RDD，映射到Row上去val textb = spark.sparkContext.textFile("E:\\若泽数据\\零基础大数据篇第三期\\Hadoop综合编程\\b.txt").filter(_ != "").map(_.split(" ")).map(x => Row(x(0).trim, x(1).trim))val fields = types.StructType(List(StructField("url", StringType, true),StructField("user", StringType, true) //StructField字面翻译不就是字段结构化嘛))val schematype = StructType(fields) //这段代码的作用是啥呢?放不放都不影响结果val infoDFa = spark.createDataFrame(texta, schematype)infoDFa.createOrReplaceTempView("INFOA")val infoDFb = spark.createDataFrame(textb, schematype)infoDFb.createOrReplaceTempView("INFOB")//TODO...(1)求俩个文件的IP数目（需不需要考虑重复的情况呢）spark.sql("select count(distinct url) from INFOA").show(false)spark.sql("select count(distinct url) from INFOB").show(false)//TODO...（2）求出现在b.text而没有出现在a.text的IP（查询a里面没有，b里面有的数据） spark.sql("select * from INFOB where " +"(select count(url) as num from INFOA where INFOB.url = INFOA.url)= 0").show(false)/* 其实1就代表你这个查询的表里的第一个字段这里用1，也是为了方便，当然如果数据量较大的话，也可以提高速度，因为写count(*)的话会所有列扫描，这里用1的话或者用字段名的话，只扫描你写的那个列count(*)和count(字段名)  基本结果是一样的但是一种情况例外，就是当某字段名下边的数据有null值的时候，不计入这个count中，*则全部列入count中
*//* spark.sql("select distinct INFOB.url from  INFOA where " +"INFOB.url not in (select url from INFOA)")//在表中，可能会包含重复值。这并不成问题，不过，有时您也许希望仅仅列出不同（distinct）的值。关键词 distinct用于返回唯一不同的值。//select distinct name, id from A  就是去重的一个方法*//* spark.sql("select INFOA.url from INFOA left join INFOB on INFOA.url=IFNOB.url where INFOB.url is null").show(false)infoDFb.join(infoDFa,infoDFa.col("url")===infoDFb.col("url")).show()*///TODO...(3)每个user出现的次数和每个user对应的url个数//求的是每个user出现的次数spark.sql("SELECT  user,COUNT(*)  FROM INFOA GROUP BY user").show(false)//求每个user对应的ip数spark.sql("SELECT  user,COUNT(url)  FROM INFOA GROUP BY user").show(false)/*    spark.sql("SELECT  user,url,COUNT(url)  FROM INFOA GROUP BY user,url").show(false)*/// spark.sql("SELECT COUNT(url)  FROM INFOA where user='zhangsan'").show(false)//TODO...(4)对应IP数最多的前K个user(user,url)spark.sql("SELECT  user,COUNT(url)  FROM INFOA GROUP BY user order by COUNT(url) desc limit 10").show(false)//select top 10 user,count(url)  from INFOA  group by user order by  count(url) desc/*select count(*) as ct,a.typeid as cc,b.typename from equip_product a inner join protype b on(a.typeid=b.id) group by a.typeid order by 1 desc*/spark.stop()}}