关于Spark的面试题，你应该知道这些！

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

    主要功能：管理当前节点内存，CPU的使用状况，接收master分配过来的资源指令，通过ExecutorRunner启动程序分配任务。

    worker就类似于包工头，管理分配新进程，做计算的服务，相当于process服务。

    需要注意的是：

    1）worker会不会汇报当前信息给master？**worker心跳给master主要只有workid，它不会发送资源信息以心跳的方式给master**，master分配的时候就知道work，只有出现故障的时候才会发送资源。

    2）worker不会运行代码，具体运行的是Executor是可以运行具体appliaction写的业务逻辑代码，操作代码的节点，它不会运行程序的代码的。

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

    RDD（Resilient Distributed Dataset）叫做分布式数据集，是spark中最基本的数据抽象，它代表一个不可变，可分区，里面的元素可以并行计算的集合。

    五大特性：

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

    具体做法可以在会产生大量对象的map操作之前调用repartition方法，分区成更小的块传入map。

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

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

    这种情况的解决方法就是同时配置–executor-cores或者spark.executor.cores参数，确保Executor资源分配均匀。使用rdd.persist(StorageLevel.MEMORY\_AND\_DISK\_SER)代替rdd.cache()。

    cache：缓存数据，默认是缓存在内存中，其本质还是调用persist。

    persist:缓存数据，有丰富的数据缓存策略。数据可以保存在内存也可以保存在磁盘中，使用的时候指定对应的缓存级别就可以了。

    这个常出现在笔试阶段，手写WordCount算是一项基本技能。

//创建SparkConf并设置App名称和master地址
val conf=new SparkConf().setAppName(“wc”).setMaster(“Local[*]”)
//创建SparkContext，该对象是提交Spark App的入口
val sc=new SparkContext(conf)
//使用sc创建RDD并执行相应的transformation和action
val result=sc.textFile(“输入文件的路径”)
Val rdd2=result.flatmap(x=>x.split(“ ”))
.map((_,1)).reduceBykey((_+_)).saveAsTextFile(“输出文件路径”)
//关闭链接
sc.stop()

    1、从集合中创建RDD；

val rdd = sc.parallelize(Array(1,2,3,4,5,6,7,8))
val rdd = sc.makeRDD(Array(1,2,3,4,5,6,7,8))

    2、从外部存储创建RDD；

val rdd= sc.textFile("hdfs://node01:8020/data/test")

    3、从其他RDD创建。

val rdd1 = sc.parallelize(Array(1,2,3,4))
val rdd2 =rdd.map(x=>x.map(_*2))

    这个涉及到的算子就比较多了，感兴趣的朋友可以去看看博主的这两篇博客:

    绝对不会让你失望的~

    窄依赖指的是每一个父RDD的Partition最多被子RDD的一个Partition使用。

     宽依赖指的是多个子RDD的Partition会依赖同一个父RDD的Partition，会引起shuffle。

    RDD任务切分中间分为：Application、Job、Stage和Task

    1）Application：初始化一个SparkContext即生成一个Application；

    2）Job：一个Action算子就会生成一个Job；

    3）Stage：根据RDD之间的依赖关系的不同将Job划分成不同的Stage，遇到一个宽依赖则划分一个Stage；

    4）Task：Stage是一个TaskSet，将Stage划分的结果发送到不同的Executor执行即为一个Task

    弹性分布式数据集；不可变、可分区、元素可以并行计算的集合。

    DataFrame以RDD为基础的分布式数据集。

    DataSet包含了DataFrame的功能，Spark2.0中两者统一，DataFrame表示为DataSet[Row]，即DataSet的子集。

    （1）DataSet可以在编译时检查类型；

    （2）并且是面向对象的编程接口。

    （DataSet 结合了 RDD 和 DataFrame 的优点，并带来的一个新的概念 Encoder。当序列化数据时，Encoder 产生字节码与 off-heap 进行交互，能够达到按需访问数据的效果，而不用反序列化整个对象。）。

    1）创建DataFrame

scala> val df = spark.read.json("/export/spark/examples/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

    2）打印数据

scala> df.show()
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|

    3）注册UDF，功能为在数据前添加字符串

scala> spark.udf.register("addName", (x:String)=> "Name:"+x)
res5: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))

    4）创建临时表

scala> df.createOrReplaceTempView("people")

    5）应用UDF

scala> spark.sql("Select addName(name), age from people").show()
+-----------------+----+
|UDF:addName(name)| age|
+-----------------+----+
|     Name:Michael|null|
|        Name:Andy|  30|
|      Name:Justin|  19|

    本篇博客就分享到这里，建议所有没看过开头提到的[《不会这20个Spark热门技术点，你敢出去面试大数据吗?》](https://blog.csdn.net/weixin_44318830/article/details/106337941)这篇博客的朋友都去阅读一下，真的墙裂推荐!!!