专注大数据培训
我们一直在领跑

最新面试题集锦(10)- 汇纳科技大数据开发笔试题

1.Spark on Yarn作业执行流程?yarn-client和yarn-cluster有什么区别?

Spark on Yarn作业执行流程?

1.Spark Yarn Client 向 Yarn 中提交应用程序。
2.ResourceManager 收到请求后,在集群中选择一个 NodeManager,并为该应用程序分配一个 Container,在这个 Container 中启动应用程序的 ApplicationMaster, ApplicationMaster 进行 SparkContext 等的初始化。
3.ApplicationMaster 向 ResourceManager 注册,这样用户可以直接通过 ResourceManager 查看应用程序的运行状态,然后它将采用轮询的方式通过RPC协议为各个任务申请资源,并监控它们的运行状态直到运行结束。
4.ApplicationMaster 申请到资源(也就是Container)后,便与对应的 NodeManager 通信,并在获得的 Container 中启动 CoarseGrainedExecutorBackend,启动后会向 ApplicationMaster 中的 SparkContext 注册并申请 Task。
5.ApplicationMaster 中的 SparkContext 分配 Task 给 CoarseGrainedExecutorBackend 执行,CoarseGrainedExecutorBackend 运行 Task 并向ApplicationMaster 汇报运行的状态和进度,以让 ApplicationMaster 随时掌握各个任务的运行状态,从而可以在任务失败时重新启动任务。
6.应用程序运行完成后,ApplicationMaster 向 ResourceManager申请注销并关闭自己。

yarn-client和yarn-cluster有什么区别?

1.理解YARN-Client和YARN-Cluster深层次的区别之前先清楚一个概念:Application Master。在YARN中,每个Application实例都有一个ApplicationMaster进程,它是Application启动的第一个容器。它负责和ResourceManager打交道并请求资源,获取资源之后告诉NodeManager为其启动Container。从深层次的含义讲YARN-Cluster和YARN-Client模式的区别其实就是ApplicationMaster进程的区别 2. YARN-Cluster模式下,Driver运行在AM(Application Master)中,它负责向YARN申请资源,并监督作业的运行状况。当用户提交了作业之后,就可以关掉Client,作业会继续在YARN上运行,因而YARN-Cluster模式不适合运行交互类型的作业 3. YARN-Client模式下,Application Master仅仅向YARN请求Executor,Client会和请求的Container通信来调度他们工作,也就是说Client不能离开

2.Transfoemation和Action是什么?有什么区别?列举几个常用的方法?

spark支持两种RDD的操作:transformation和action。transformation操作会针对已有的RDD创建一个新的RDD;而action则主要对RDD进行最后的操作,比如遍历,保存到文件等等,并可以返回结果给Driver程序。 ①transformation的特点就是lazy特性。lazy特性指的是,如果一个spark应用中只定义了transformation操作,那么即使你执行该应用,这些操作也不会执行。也就是说,transformation是不会触发spark程序的执行的,它们只是记录了对RDD所做的操作,但是不会自发的执行。只有当transformation之后,接着执行了一个action操作,那么所有的transformation才会执行。Spark通过这种lazy特性,来进行底层的spark应用执行的优化,避免产生过多中间结果。 ②action操作执行,会触发一个spark job的运行,从而触发这个action之前所有的transformation的执行。这是action的特性。

常用的transformation的算子:

1、sortBy : 对于RDD的是非元组型,rdd1.soreBy(x=>x),元组型按value排序
rdd.soreBy(_._2)(进行shuffle)

2、sortByKey : 对于RDD内存的是元组类型的使用(进行shuffle) ,sortBy和
sortByKey都是transformation算子,但都会触发job任务,底层new了一个rangePartition对象,底层会调用一个foreach方法,进行排序,所以会触发job

3、reduceByKey(+) : 先在分区内进行按key合并,在全局合并(全局是把相同key的
不同的分区,拉倒同一个分区(有多个分区))会进行shuffle

4、filter: 过滤, 不会进行shuffle

5、flatMap: 方法,可以先切分,再压平, 不会进行shuffle

6、 rdd1.intersection(rdd2), 求两个集合的交集

7.0、 rdd1.union(rdd2),返回rdd1和rdd2中的所有元素,返回类型是Array,不Shuffle

7、rdd1.join(rdd2), 返回结果RDD[(String, (Int, Int))],join是内连接,
只有相同的才会join,会有shuffle过程

8、 rdd1.leftOuterJoin(rdd2),返回结果是RDD[(String, (Int, Option[Int]))],如果有rdd2中有和rdd1对应的数据时Some(value值),没有值None,第一个Int是rdd1中的value,rdd1中的数据会显示完,有shuffle

9、rdd1.rightOuterJoin(rdd2),返回结果是RDD[(String, (Option[Int], Int))],
String是key类型, Option[Int]表示和rdd2中对应的没有,就和None,
有和rdd2中的数据对应的就是Some(value),rdd2中的数据会显示完,有shuffle

10、rdd.groupByKey(), 按key进行聚合,把不同分区的相同的key拉倒同一台机器上,返回值是 RDD[(String, Iterable[Int])],String 是key的类型,迭代器是相同key的value的,如果想要聚合 ,可以调用 map(x=>(x._1,x._2.sum))方法,相当于reduceBykey.会进行shuffle,
10.1 rdd.groupBy() 通过指定的值进行聚合,返回的值是
RDD[(String, Iterable[(String, Int)])] ,迭代器中的类型是整个KV的类型

11、rdd.reduceByKey(+),返回值RDD[(String, Int)],在分区内按key进行聚合
(Combiner),再把不同分区的相同的key的数据拉倒同一机器上进行聚合,会进行shuffle,
reduceByKey会进行Combiner,所以比groupByKey效率高,一般用reduceByKey

常用的action算子:

1、collect 无shuffle:返回的是Array类型,打印的话需要toBuffer
2、sum 无shuffle:返回值是double类型
3、reduce(+) :可以用于替换sum, 返回值与调用的值有关
4、rdd.countByKey()有shuffle,计算相同的key出现的次数,与value无关,返回值是collection.Map[String, Long], String是指的是Key,Long指的是相同key出现的次数。底层实现是reduceByKey,
def countByKey(): Map[K, Long] = self.withScope {
self.mapValues(_ => 1L).reduceByKey(_ + _).collect().toMap
},把value都换成1,在求相同key 的和
5、count ,计数
6、take,可以用于求topN, 获取指定的几条数据
7、foreach
8、saveAsTextFile
9、foreachPartition

3.什么是数据倾斜?(spark任务)为什么会出现数据倾斜?如何解决?

什么是数据倾斜

数据倾斜指的是,并行处理的数据集中,某一部分(如spark或者kafka的一个partition)的数据显著多于其他部分,从而使得该部分的处理速度成为整个数据集处理的瓶颈。

为什么会出现数据倾斜

在Spark中,同一个Stage的不同Partition可以并行处理,而具有依赖关系的不同Stage之间是串行处理的。假设某个Spark Job分为Stage 0和Stage 1两个Stage,且Stage 1依赖于Stage 0,那Stage 0完全处理结束之前不会处理Stage 1。而Stage 0可能包含N个Task,这N个Task可以并行进行。如果其中N-1个Task都在10秒内完成,而另外一个Task却耗时1分钟,那该Stage的总时间至少为1分钟。换句话说,一个Stage所耗费的时间,主要由最慢的那个Task决定。
由于同一个Stage内的所有Task执行相同的计算,在排除不同计算节点计算能力差异的前提下,不同Task之间耗时的差异主要由该Task所处理的数据量决定。

如何解决数据倾斜?

①调整并行度分散同一个Task不同Key Spark在做Shuffle时,默认使用HashPartitioner(非Hash Shuffle)对数据进行分区。如果并行度设置的不合适,可能造成大量不相同的Key对应的数据被分配到了同一个Task上,造成该Task所处理的数据远大于其它Task,从而造成数据倾斜。
如果调整Shuffle时的并行度,使得原本被分配到同一Task的不同Key发配到不同Task上处理,则可降低原Task所需处理的数据量,从而缓解数据倾斜问题造成的短板效应。 用场景
大量不同的Key被分配到了相同的Task造成该Task数据量过大。 解决方案
调整并行度。一般是增大并行度,但有时如本例减小并行度也可达到效果。 优势
实现简单,可在需要Shuffle的操作算子上直接设置并行度或者使用spark.default.parallelism设置。如果是Spark SQL,还可通过SET spark.sql.shuffle.partitions=[num_tasks]设置并行度。可用最小的代价解决问题。一般如果出现数据倾斜,都可以通过这种方法先试验几次,如果问题未解决,再尝试其它方法。 劣势
适用场景少,只能将分配到同一Task的不同Key分散开,但对于同一Key倾斜严重的情况该方法并不适用。并且该方法一般只能缓解数据倾斜,没有彻底消除问题。从实践经验来看,其效果一般。

②自定义Partitioner 使用自定义的Partitioner(默认为HashPartitioner),将原本被分配到同一个Task的不同Key分配到不同Task。 适用场景
大量不同的Key被分配到了相同的Task造成该Task数据量过大。 解决方案
使用自定义的Partitioner实现类代替默认的HashPartitioner,尽量将所有不同的Key均匀分配到不同的Task中。 优势
不影响原有的并行度设计。如果改变并行度,后续Stage的并行度也会默认改变,可能会影响后续Stage。 劣势
适用场景有限,只能将不同Key分散开,对于同一Key对应数据集非常大的场景不适用。效果与调整并行度类似,只能缓解数据倾斜而不能完全消除数据倾斜。而且需要根据数据特点自定义专用的Partitioner,不够灵活。

③将Reduce side Join转变为Map side Join 通过Spark的Broadcast机制,将Reduce侧Join转化为Map侧Join,避免Shuffle从而完全消除Shuffle带来的数据倾斜。 适用场景
参与Join的一边数据集足够小,可被加载进Driver并通过Broadcast方法广播到各个Executor中。
优势
避免了Shuffle,彻底消除了数据倾斜产生的条件,可极大提升性能。
劣势
要求参与Join的一侧数据集足够小,并且主要适用于Join的场景,不适合聚合的场景,适用条件有限。

④为skew的key增加随机前/后缀 为数据量特别大的Key增加随机前/后缀,使得原来Key相同的数据变为Key不相同的数据,从而使倾斜的数据集分散到不同的Task中,彻底解决数据倾斜问题。Join另一则的数据中,与倾斜Key对应的部分数据,与随机前缀集作笛卡尔乘积,从而保证无论数据倾斜侧倾斜Key如何加前缀,都能与之正常Join。

4.Hive内部表、外部表与分区表、桶表的区别?

内部表:又叫管理表,表的创建,和删除都由hive自己决定。
外部表:表结构上同内部表,但是存储的数据时自己定义的,外部表在删除的时候只删除元数据,原始数据时不能删除的。

内部表和外部表的区别主要体现在两个方面:

  • 删除:删除内部表,删除元数据和数据;删除外部表,删除元数据,保留数据。
  • 使用:如果数据的所有处理都在 Hive 中进行,那么倾向于 选择内部表,但是如果 Hive 和其他工具要针对相同的数据集进行处理,外部表更合适。使用外部表访问存储在hdfs上的数据,然后通过hive转化数据并存储到内部表中。

hive中分桶表和分区表的区别

分区表: 原来的一个大表存储的时候分成不同的数据目录进行存储。
  如果说是单分区表,那么在表的目录下就只有一级子目录,如果说是多分区表,那么在表的目录下有多少分区就有多少级子目录。不管是单分区表,还是多分区表,在表的目录下,和非最终分区目录下是不能之间存储数据文件的 分桶表: 原理和hashpartitioner 一样,将hive中的一张表的数据进行归纳分类的时候,归纳分类规则就是hashpartitioner。(需要指定分桶字段,指定分成多少桶)

分区表和分桶的区别除了存储的格式不同外,最主要的是作用:

  • 分区表:细化数据管理,缩小mapreduce程序 需要扫描的数据量
  • 分桶表:提高join查询的效率,在一份数据会被经常用来做连接查询的时候建立分桶,分桶字段就是连接字段;提高采样的效率

5.如何避免读、写HBaes时访问热点问题?

(1)加盐
这里所说的加盐不是密码学中的加盐,而是在rowkey的前面增加随机数,具体就是给rowkey分配一个随机前缀以使得它和之前的rowkey的开头不同。给多少个前缀? 这个数量应该和我们想要分散数据到不同的region的数量一致(类似hive里面的分桶)。
( 自己理解: 即region数量是一个范围,我们给rowkey分配一个随机数,前缀(随机数)的范围是region的数量)
加盐之后的rowkey就会根据随机生成的前缀分散到各个region上,以避免热点。

(2)哈希
哈希会使同一行永远用一个前缀加盐。哈希也可以使负载分散到整个集群,但是读却是可以预测的。使用确定的哈希可以让客户端重构完整的rowkey,可以使用get操作准确获取某一个行数据。

(3)反转
第三种防止热点的方法是反转固定长度或者数字格式的rowkey。这样可以使得rowkey中经常改变的部分(最没有意义的部分)放在前面。这样可以有效的随机rowkey,但是牺牲了rowkey的有序性。反转rowkey的例子:以手机号为rowkey,可以将手机号反转后的字符串作为rowkey,从而避免诸如139、158之类的固定号码开头导 致的热点问题。

(4)时间戳反转
一个常见的数据处理问题是快速获取数据的最近版本,使用反转的时间戳作为rowkey的一部分对这个问题十分有用,可以用Long.Max_Value – timestamp追加到key的末尾,例如[key][reverse_timestamp] ,[key] 的最新值可以通过scan [key]获得[key]的第一条记录,因为HBase中rowkey是有序的,第一条记录是最后录入的数据。

(5)尽量减少行和列的大小
在HBase中,value永远和它的key一起传输的。当具体的值在系统间传输时,它的rowkey,列名,时间戳也会一起传输。如果你的rowkey和列名很大,HBase storefiles中的索引(有助于随机访问)会占据HBase分配的大量内存,因为具体的值和它的key很大。可以增加block大小使得storefiles索引再更大的时间间隔增加,或者修改表的模式以减小rowkey和列名的大小。压缩也有助于更大的索引。

(6)其他办法
列族名的长度尽可能小,最好是只有一个字符。冗长的属性名虽然可读性好,但是更短的属性名存储在HBase中会更好。也可以在建表时预估数据规模,预留region数量,例如create ‘myspace:mytable’, SPLITS => [01,02,03,,…99]

编程题:

有大量英文文献,需要统计以下信息,请编程实现(编程语言不限)

(1)统计出现频次最高的前100个单词

(2)统计出现1-10次,11-20次,21-50次,51次以上各区间的单词数量。

val spark = SparkSession.builder().appName("demo").getOrCreate()
    import spark.implicits._
    val docs = spark.read.textFile("/path")
    val words = docs.flatMap(line=>{
      line.split(" ")
    }).toDF("word")


    words.createTempView("words")


   val wordcount =  spark.sql(
      """
        |
        |select
        |word,count(1) as cnt
        |
        |from words
        |group by words
        |
        |""".stripMargin)

    // 次数top100 词
    val top100 = wordcount.limit(100)

    wordcount.createTempView("wordcount")
    // 区间词数
    val rangeWords = spark.sql(
      """
        |
        |select
        |rg,
        |count(1) as words
        |from
        |(
        |
        |select
        |
        |word,
        |case
        | when cnt between 1 and 10 then 'range1'
        | when cnt between 11 and 20 then 'range2'
        | when cnt between 21 and 50 then 'range3'
        | else 'range4'
        |end as rg
        |
        |
        |from wordcount
        |)
        |group by rg
        |
        |
        |""".stripMargin)

    spark.close()

  }
}
欢迎分享,转载有奖:多易教育 » 最新面试题集锦(10)- 汇纳科技大数据开发笔试题