黄伟佳面试总结_scala创建的redis连接池能在mappartition里调用吗

面试自我介绍

2.Hadoop
HDAOOP:是一个分布式系统的基础架构,其优点
高可靠:就是其会存储多个副本
高扩展:是可以拓展数以千计的节点
高效性:是因为其并行处理任务会大大的加快任务处理完成的速度
高容错:能够将失败的任务重新分配
HDAOOP:组件我了解到的大概有如下这些HDFS,HIVE,HBASE,YARN,SPARK,ZK,MR
HDFS:为一个分布式的文件存储系统:他是一个高容错,并且能检测和应对硬件故障的文件存储系统,
HDFS工作机制:
NameNode会将客户端上传的数据根据默认的大小进行切分后存储到指定的DataNode上,并且一个文件块会有多个副本,可以用来确保数据的安全,即使一台机器挂掉或者数据损坏也可以从其他的DataNode进行读取,并且DataNode的定期的向NameNode就行汇报自身的文件块信息,而NameNode则会确保文件块副本的数量.
HDFS读取文件流程:客户端先向NameNode进行访问元数据,获取到数据所在的DataNode,后和DataNode就行通信获取到所需要的数据
HDFS写数据的流程:客户端先向NameNode发送请求后,NameNode进行检查其文件是否存在,及其父目录是否也存在,以及用户是否有权限写入数据,客户端接收到可以上传的请求后再次请求上传数据块,NameNode接收到请求后返回可以存储数据块的DataNode,后客户端开始上传数据,然后第一台上传给第二台后第二台传给第三台,然后逐级上传成功后,客户端再次上传第二个数据块,直至数据全部上传完成.
HDFS中一个DataNode宕机怎么一个流程恢复:
短暂的宕机可以写脚本进行重启,如果长时间宕机则会将这台机器的数据备份到另一个机器上后,清空数据重启
HDFS中namenode宕机怎么处理:
可以将secondyNameNode节点的数据拷贝到备用Namenode上这样就可以恢复在上一个edits和fsimage的内容了
存储格式:
Sequencflie MapFlie RCFile ORC Parquet,我们常用的为ORC Parquent
压缩格式:
Gzip LZO Snappy Bzip2 我们常用Gzip 或者lzo
HBASE:是一个分布式的列存储的数据库,其主要就是为了方便用户进行实时的读写数据以及随机访问超大规模的数据集:
其写流程个人理解如下:clinet连接zk找到hbase meta表所在的region server,后region server扫描meta表获取到插入数据所在的region范围,找到对应负责的regionserver ,然后客户端向其发生请求,写入数据
读数据的话,就会先通过zk访问,获取meta表所在的regionserver,对应的regionserver,扫描meta表,找到查询的key在那个region 该region由那个regionserver负责后,发送请求,返回数据,会现在memstore中找数据,如果没有就会在Blockcache当中读取,如果在没有就到store file上读取
Rowkey原则的话主要是:散列原则,长度原则,唯一原则,我之前设计rowkey的话会将订单id和long的最大值减去时间的时间戳来进行使用,这样的话可以通过订单id避免region的热点问题,以及可以按时间倒排进行显示
为什么会快:因为它使用的是LSM树形结构,而不是b或者b+树,而且hbase读取数据的话会先读取缓存当中,如果缓存当中没有再回从内存中的memstore中查找.如果都找不到才会加载hFlie中的内容,并且也会很快,因为节省了省道开销
3.Hive:
Hive:最初用于解决海量结构化的日志数据统计问题.hive定义了一种类似sql的查询语言（hql）将sql转化为map reduce任务在hadoop上执行。
1.1.Hive原理:
解析器将sql字符串转化成抽象的语法树，遍历语法树生成逻辑执行计划，优化器对执行计划进行优化，然后执行器将优化后的逻辑计划翻译成map reduce任务。
1.2.Hive执行流程?
客户端或者命令行提交hql语句,在driver中通过解析器将其转换为一个抽象的语法树,后经编译器将抽象的语法树转换为一个逻辑计划,存储在hdfs中,后经过优化器进行优化,由执行器调用底层执行框架执行逻辑计划生成MR运行在HADOOP集群中
1.3.Hive文件格式:
我们一般使用parquent或者Orc格式,还有就是rc text,seqfile格式,但是不建议使用
1.4.Hive数据倾斜优化方案:
数据倾斜的本质就是数据分配的不均匀,那么造成不均匀的原因有很多,比如热点key,非热点key分布不均匀的等问题,那么需要定位到具体位置发生原因才能对症下药,从技术层面来说,首先怀疑是否是数据存在热点key的问题.这里可以采样统计来判断是否有部分key记录过多,如果是热点key则可以采用单独处理或者二次聚合的方式来处理,如果不是热点key,那么有可能就是key的本身分配存在问题,那么可以选择增加随机前缀,使用group代替distinct mapjoin 空值等一些手段将key进行随机分区,那如果是业务层面来说可能是提供计算口径有问题导致的数据倾斜,比如订单表和支付关联,正常100w从结果来看确是1000w或者更多,这样就需要考虑是否逻辑口径有问题了,以上两种方案,然后再结合一些参数优化就可以解决倾斜问题
1.5.Hive 大表join 小表产生的问题,怎么解决?
容易产生数据倾斜.解决的话就是将空值的key代替为随机值,也可以将热点数据和非热点数据分开处理,最后在进行合并,然后也可以使用gruop by代替distinct
1.6.Hive如何合并小文件:
可以在Map输入时合并小文件,这里可以配置,在Reduce输出的时候合并小文件
Hive会在正常job结束后起一个 task来判断是否需要合并小文件,如果满足就会另外启动一个map-only job或者mapred job来完成小文件合并,这就需要配置一些参数
1.7.Hive中 udf udaf udtf 中有什么区别怎么实现?
三者均为自定义函数
udf简化来说就是一对一的关系 就是用户自定义函数,是输入一行数据返回也是一行数据的自定义算子
Udaf简化来说就是多对一关系 ,就是用户自定义聚合函数,是输入多行函数返回一行数据
Udtf简化来说就是一对多的关系,就是用户自定义生成函数,是输入一行返回多行函数又被称为表生成函数
具体实现方式均需要继承对应的接口后实现里面的方法:
1.8.Hive元数据保存方式:
Hive元数据有三种方式保存元数据,分别为内嵌式,本地式,远程式
内嵌式则是将元数据能访问一个数据文件,不支持多会话连接保存到内嵌的Derby数据库当中,每次只
本地式则是将数据存入到本地独立的数据库当中,可支持多会话连接,
远程式则是将数据保存到远程独立的数据库当中,这样就避免每个客户端都去安装MqSQL
1.9.Hive四个表的区别:
内部表和外部表都是基本表，不同点就是数据的位置，在Hive中进行表的删除时内部表的数据会被一并删除。当外部表删除时，其所对应的数据文件不会被删除。故生成环境下一般都使用外部表,这样 并不会删除数据
分区表是将数据按照分区字段拆分存储的表,在hdfs中以文件夹的形式分别存放不同的分区数据
分桶表则是将分桶字段hash值分组拆分数据的表,在HDFS中表现为将单个数据拆分多个文件
分桶和分区的区别;表现形式的话分桶是文件,分区是文件夹,创建语句上的话分区是partition by 分桶是cluster by
数量上分区可增长,而分桶不可增长,作用上的话,分区只是避免全局扫描,分桶表会保持分桶查询结果到桶结构,进行join可以提高效率.
两者可结合使用,来方便数据的查询效率
1.10.insert into 和 override write的区别:
Insert into只是单纯的插入数据,返回会将历史数据和新插入的一起返回
Override write 会将历史数据remove后再插入数据;
1.11.Hive中的判断函数:
If 返回结果为true或者False
nvl:判断两值那个不为空
case when 类似于if else的多重判断
coalesce返回第一个null值,可以添加多个参数
Is null判断是不是空值
Is not null判断否不是空值
1.12.Hive的创表语句:
建表语句的话有三种,
第一种:基础的创表语句也就是直接建表法,create table 表名(列名 属性 )row format delimited fields terminated by’分隔符’
第二种:like方式,create table 表 like表 复制表格式不包含数据
第三种:子查询方式(抽取原表的部分字段分析数据) create table 表 as select 列名from原表名
1.13.线上每天产生>=3g的业务日志,每天需要加载到hive的log表中,将每天的业务日志压缩后load到hive表中,最好使用那个压缩算法,说明原因?
我司通常使用snappy算法,原因是Hive的话会比较支持ORC的格式存储,因为orc的格式也是列式存储的,然后两者结合起来会更好一些,后面我了解到可以使用lzo这个因为该压缩算法可以切分,比率高,解压快非常适合日志
1.14.Union all 和union的区别?
union 去掉重复数据
union all 不去掉重复数据
1.15.说一下印象最深的一次优化场景,hive常见优化思路:
一方面可以根据存储进行优化,这里的话可以将压缩格式改变一下,也可以将小文件进行优化,另一方面就是计算方面的优化,这里的话可以进行本地化执行,JVM重用,调解MAP和Reducer个数,并行度的设置,再或者就是优化一些sql,或者是调解一下内存的分配问题
1.16.简述delete，drop，truncate的区别:
delete 删除行数据和整个表数据进行删除
drop 删除表内容和定义以及表结构也会删除
Truncate 摧毁表结构并重建
1.17Hive排序的区别;
Order by 全局排序 需要shuffle ,partitionBy(分区)后面跟orderby
Sort by 部分排序(及每个reducer内排序 全局不一定有序)
Distribute by 分区排序类似MR中的partition 后使用Sort by
Cluster by    当distribute by 和 sort by 所指定的字段相同时，即可以使用cluster by。
Group by 按照指定字段进行分组,相同的key到相同reduce 后续是聚合操作
1.17.Hive 里边字段的分隔符用的什么？为什么用\t？有遇到过字段里 边有\t 的情况吗，怎么处理的？
Hive默认字段的分隔符为ascii码中的控制符\001,建表时用fields terminated by ‘\001’,遇到\t的情况自定义InputForamt,替换其他分隔符在处理
1.18.Hive中MapJoin的原理:
大概流程就是:当发生MapJoin的时候,小表会被复制分发到Map端,并加载到内存当中,然后顺序扫描大表根据key在内存中依次进行关联join,避免shuffle过程,但是需要注意map join没有reduce,所有文件的数量和map的数量有关,这里要注意小文件的问题
1.19.Hive的row_number中distributer by和partition by的区别?
row_number() over( partition by 分组的字段 order by 排序的字段)
row_number() over( distribute by 分组的字段 sort by 排序的字段)
注意:
partition by 只能和order by 组合使用
distribute by 只能和 sort by 组合使用
1.20.Hive开发中遇到什么困难?
存储格式问题Parquet不支持Date
数据过滤的问题 sql做left join 操作时，on两边都是null值时，所有数据被过滤
当做join操作时，如果出现null值，可能会导致没有数据，此时需要将null字段设置为默认值
小文件的问题,需要将其合并会在控制小文件的产生,以及有可能查收中文乱码的问题
1.21.Hive常用函数有哪些函数?
函数的话可以分为两大类,一类为内置函数,一类为自定义函数:其中内置函数又分为数学函数,收集函数,类型转换函数,日期函数,条件函数,字符函数 内置的聚合函数,内置的表生成函数,
其中常用的有:
size 返回map或者数组中的数据类型
Cast 类型转换以免后续计算出现异常
Datediff 时间差
Date_add 添加时间
Date_sub 减少时间
Uninx_timestamp 返回日期的毫秒值
If 判断是否满条件
Cocat 字符串的合并
Count 返回条数
Sum 求和
Avg 求平均值
Hive 列转行,把某一列的数据，在一行中展示。 lateral view explode
select *from 表 lateral view explode(split(列,‘切分’)) 表名 as 列名
Hive 行转列 使用函数：concat_ws(’,’,collect_set(column))
说明：collect_list 不去重，collect_set 去重。 column 的数据类型要求是 string
select 列,concat_ws(’拼接符’,collect_list(列名)) as 别名 from 表名 group by 字段;
1.22.手写sql 连续活跃:
– 第二段 根据登陆时间减去排列顺序 得出用户连续登陆的分组
SELECT user_id,date_sub(login_time,rank) as login_group,min(login_time) as start_login_time
,max(login_time) as end_login_time,count(1) as continuous_days
FROM (
– 第一段首先根据用户分组，登陆时间排序，结果按照登陆时间升序排列
SELECT user_id,login_time,row_number() OVER(PARTITION BY user_id order by login_time) as rank
FROM test.user_login
) a
GROUP BY user_id,date_sub(login_time,rank)
1.23.Hive 中 left semi join 和 left join的区别?
Left semi join只显示左表的信息,在select后面不能出现查询右表的字段,否则报错,where条件中也不可以有右表字段出现 而left join 可以显示右表字段,并且where条件中也可以有右表字段出现 并且where中出现了右表字段则等同于join 和inner join
Left semi join 如果右边有重复的数据,对应的左表没有重复数据,那么就不会有重复的数据产生,如果两表都有则会和左表一样
而 left join 的右表有重复的数据对应左表没有则会产生重复数据,如果两表都有重复则查出的结果为两边的乘积
Left join 时where没有右表字段,那么当右表的数据不存在而左表存在,那么之后返回左表的信息
Left semi join 当右表中不存在时,不显示任何信息
1.24.Hive 的执行引擎,spark和MR的区别?
Spark的话是基于内存进行计算的,效率会略快 而MR是基于磁盘进行计算的效率会略慢一些,但是对于大量数据Hive on spark 可能会出现内存溢出的问题
1.25.Hive中的join底层MR是怎么实现的?
Join的话分为两种join一种是mapjoin这种只适合与一张小表去join一张大表的情况下,而且还会有一定的优化,大体分为两步先是将小表的数据缓存到内存中的哈希表中,读完后会将哈希表序列化为哈希文件,然后会分发的到每个Mapper的本地磁盘上进行join,这样就会减少shuffle以及reduce的操作,但是局限于只支持小表join大表的情况.

common join这种就是在没有更改配置文件以及不符合上述规则情况下进行join,整个过程包含map shuffle 以及reduce阶段.在map端输出的时候以 join on条件中的列为key,如果join多个关联键则以这些关联键组合为key , shuffle阶段是将key进行哈希值后分发给不同的reduce, reduce阶段根据key进行join,并且通过tag的信息标明数据来源.
1.26.建好外部表以后用什么语句将数据加载到表中?
数据在本地的话:load data local inpath本地路径 into table 表名
数据在hdfs的话:load data inpath hdfs路径+文件 into表名
1.27.Hive和传统的数据库有什么区别?
查询语言的区别:HiveSQL是HQL语言,传统SQL是SQL语句:
数据存储的位置不同:HIVE的数据存储在Hdfs中,而传统数据库存储在块设备或者是本地文件
数据格式的不同:hive数据格式可以用户自定义,而传统sql有系统自己的定义格式
数据更新:hive不支持数据的写操作,而sql可以支持数据更新
索引区别:hive中没有索引,而sql有索引
延迟性区别:Hive延迟性率高,而mysql延迟性低
数据规模区别:hive存储规模超级大,而mysql存储一些少量数据
底层执行原理区别:HIve底层是mr 而mysql是excutor执行器
可拓展性:hive高于传统数据库
1.28.Hive导入数据的方式
本地导入:load data local inpath本地路径 into table 表名
hdfs导入:load data inpath hdfs路径+文件 into表名
查询导入:create table表 as select * from ods.test
查询结果导入:insert into 表 tmp.test select * from ods.test
1.29.怎么将字符串中的数字拆出来?
可以通过substr来进行截取字符串后将数字输出
格式1： substr(string string, int a, int b);
格式1：
1、string 需要截取的字符串
2、a 截取字符串的开始位置（注：当a等于0或1时，都是从第一位开始截取）
3、b 要截取的字符串的长度
格式2：substr(string string, int a) ;
格式2：
1、string 需要截取的字符串
2、a 可以理解为从第a个字符开始截取后面所有的字符串。
1.33 map join的缺点
只适合进行小表join 大表,并不是很灵活
1.34大表join分桶的原因
提高效率.因为此时可以对两张大表的字段进行分桶,那join的话就是按照一个一个桶来join的 这样就完美解决了,但是两表必须确保分桶数量为倍数关系,否则会改变join条件,影响最终结果
1.35 hive去重的几种方法:
我知道大概如下几种方法.group by distinct row_number 可以取出为一的值以及collect__set是将某列转一个数组返回,并且去重
1.36hive 如何取样
大概分为三种情况:随机抽样,分桶抽样,数据块抽样
随机抽样实现的话需要使用rand函数来进行抽取,
数据块的需要使用tablesample()函数来进行取样,可以指定大小,行数.
分桶的话和数据块函数一样.但是参数会略多一些,是将rand和tablesample 组合使用
1.37row number ,rank ,dense rank 区别
三者虽然都是对分过组的数据排序加序号,三个函数都是按照分组内从1开始排序
其中,ROW_NUMBER() 是没有重复值的排序(即使两条记录相同，序号也不重复的)，不会有同名次。
DENSE_RANK() 是连续的排序，两个第二名仍然跟着第三名。
RANK() 是跳跃排序，两个第二名下来就是第四名。

1.38 hive的复杂数据类型说一下
Array(1,2,3,4) 装的数据类型是一样的
Map(‘a’,1,‘b’,c) key的类型一样
Struct(‘a’,1,2,34,4)装的数据是完全混乱的
1.38什么情况下只会产生一个reduce任务而没有maptask?
使用了udtf聚集函数却没带group by 用来order by.

1.40 有多个热点值的数据group by你们是怎么处理,优化逻辑说一下
其本质是因为数据分布不均匀，我工作的时候出现过数据倾斜，印象比较深刻的有：
前端埋点出了问题：出现大量的空key，大量的null值，选择丢弃掉。
当时的解决办法有2种，都试过：
第一种：先使用粒度比较,细的先聚合，比如先市,区聚合，然后再聚合一次，分多次聚合（直接在hive里搞就行）。
第二种：用spark代码，先把北京的取出来，在北京的前面加随机数，分到不同的task里面去，做完一次聚合后，然后再把随机数丢掉，再来一次聚合，这样速度提高10几倍，最后存到hive表里。缺点要写spark代码，略微有点麻烦。
1.41空key你们如何给随机值:
用case when给空值分配随机的 key值,使用指定字符串加随机值
1.42 where和having的区别?
用的地方不一样.where用于增删改查的所有情况,但是having只能用于查询,执行顺序也不一样
where在聚合之前起作用,但是不能放在group by 之后,而having不能放在group by之前,
然后子句也有区别,where子句的条件表达式Having都可以跟,而反之不可以;

3.Spark:快如闪电的通用计算引擎
3.1. RDD属性?
有多个分区, 分区数量决定并行度
一个功能函数作用在分区上,函数决定计算逻辑
RDD和RDD存在依赖关系,可以根据关系恢复失败的任务
分区器决定数据到那个分区当中
最优位置,即将Executor中的Task调度数据所在节点,要求Worker和DN部署在同一节点或者Onyarn,通过访问NN获取数据位置信息
3.2.算子分为那几类?(RDD支持哪几种类型操作)
大致可以分为两类,
一类为Transformation 变换/转换算子这种并不会触发提交作业并且是lazy模式,延迟执行的,
Action算子在RDD上运行计算,并返回结果给Driver或写入文件系统
3.3.分别列出几个常用的transformation 和action算子?
Transformation算子:map 将数据做函数运算 reducebykey聚合 groupbukey分组 sortbykey排序 union求并集 join
Action算子:count返回所有元素 collect将rdd的数据转换为数组 first第一条 max最大值 min最小 foreach
3.4.创建RDD的几种方式?
大致是三种创建方式.
第一种的话集合并行化创建,前提是有数据
第二种就是读取外部文件系统的如hdfs,或者是本地文件,这种也是最长用的方式,
最后一种就是从父RDD转换成新的子RDD
3.5.Spark 运行流程
启动spark context,后向资源管理器申请资源运行Executor并启动Standalone Executor Backend ,Executor向Spark context申请task, spark context将任务分发给executor, spark context构建dag, 将dag划分为stage, 将taskset发送给Taskscheduler最后由sc heduler将task发送给executor运行,运行结束后释放资源
3.6.Spark 有哪些运行模式,
本地的local模式,standalone模式,spark on yarn模式,以及spark on mesos模式其中standalone和spark on
Yarn模式,spark on mesos为集群模式
3.7.Spark中coalesce和repartition的区别
两者都是用来改变RDD的Partition数量,repartition底层调用的就是coalesce方法,
区别的话repartition一定会发生shuffle, coalesce会根据传入的参数来判断是否产生shuffle
3.8.Sort by和 Sort by key的区别?
Sort by是对标准的RDD以及KV的RDD进行排序,并且他是在spark0.9.0之后引入的,而sort by key只是对KV的RDD 进行排序
会触发action,为了构建分区规则,底层new的hashpartitioner并设置了key的排序规则,保证每一个分区中都有有序的,在全局也是有序的
3.9.Map和MapPartitions的区别
Map的话是RDD中的每一个元素进行操作
Map partition,是对RDD的每一个分区的迭代器进行操作,返回是迭代器
3.10.数据存入Redis 优先使用map map Partitions for each for each
Partition 那个?
使用for each partition
map,map partitions的话是转换类型有返回值
for each的话只可以迭代一个元素,
而使用for each partition的话他每一次迭代一个分区并且该方法没有返回值
3.11.Reduce by key 和group by key 的区别?
Reduce by key会传一个聚合函数相当于group by key+map Values,并且会有一个分区聚合,而group by key并没有
3.12.Cache和check point 的区别:
两者虽然都是持久化,但是cache是将公用的或者重复使用的RDD按照持久化级别进行优化
Check point就是将业务非常长的逻辑计算的中间结果优先缓存HDFS上,
为了效率我们可以在check point之前cache一下
3.13.Spark streaming流式统计单词数量代码

3.14.简述一下map和flat map区别和场景?
Map是对每一条函数进行指定操作,然后返回为每一条输入返回一个对象,而Flat Map函数则是两个操作的集合,即是”先映射后压平”操作和map一样,对每一条输入进行指定的操作,然后每一条返回一个对象,最后将所有对象合并为一个对象
3.15.Spark中的广播变量的用途?
使用广播变量,每个executor的内存中,只驻留一份变量副本,而不是对每个task都传输,这样就可节省网络IO提升性能
3.16.以下代码会报错么?如何解决?val arr=new ArrayList[String];
arr.foreach(parintln)
Val会报错,需将其改成val arr:Array[String]=new ArrayListString;arr.foreach就不会出现空指针异常了
3.17.说一下你用过的会产生shuffle的算子?
Distinct,去重 reducebykey 聚合 group bykey 聚合分组 aggregatebykey 按照key进行聚合 sortbykey排序算子 coalesce重分区
3.18.请写出创建dateset的几种方式?
可以直接通过create Dateset创建 然后也可以通过 case class 样类创建一个sql list ARRAY RDD 再转成dataset
3.19.描述一下RDD和DataFrame dataset的区别?
RDD的话一般不支持Spark和mlib同时使用,RDD不支持spark sql 操作
DataFrame:他和RDD以及Dataset的不同它的每一行的类型固定为row 只有通过解析才能获取各个字段的值
DataFrame和set一般与spark ml同时使用.也均支持spark sql 的操作.支持一些特别方便的保存方式
Dataset和dataframe拥有完全相同的成员函数,区别只是每一行的数据不同dataset的每一行不一样,而dataframe的为row类型的

简述:
RDD: 不支持spark sql操作
DataFrame: 每一行的固定类型为row,需要通过解析才能获取各个字段的值
DataSet: 每一行的数据不同dataset的每一行都不一样
DataFrame转RDD: testDF.rdd
DataSet转RDD: testDS.rdd
RDD转dataFrame: 需要到隐式转换, toDF(字段名)
RDD转dataSet: 隐式转换 toDS()
DataFrame转DataSet: 隐式转换 toDF.as
DataSet转DataFrame: 隐式转换 testDS.toDF

3.19.1.三者之间的互转

3.20.描述一下Spark是怎么划分stage的,描述一下shuffle?
划分思路整体就是:从后往前推,遇到宽依赖就断开,划分一个stage,遇到窄依赖就将这个RDD加入该stage
从广义上来讲就是从Map端输出数据到Reducer输入的整个过程可以广义的称为shuffle,shuffle横跨Map端和Reduce端,在Map端包括spill过程,包括输出排序,溢写合并等操作,Reduce端包括copy和sort过程,copy过程就是向各个Map任务拉取所需的数据,先回放到内存当中,达到一定程度就会输出到磁盘当中,所谓的sort就是这个合并的过程,一般都是一边copy一边sort

3.21.Spark在yarn上运行需要做那些关键的配置工作?如何kill某个Application?
这需要看是什么模式,两种模式的配置略有不同client模式需要设置指定内存,核数,而cluster模式并不需要指定
Cluster运行模式大致如下:首先客户端先向yarn提交job,然后ResouceManager 为该job在某一个NodeManager上启动ApplicationMaster,和Diver,AM启动后完成初始作业,diver生成一个DAG图,scheduler划分stage,AM向RM申请资源,RM返回Executor信息,AM通过rpc启动Executor,Executor向diver申请Task,Executor执行task,执行结果写入外部或者返回Driver端
可以通过命令行输入yarn application -kill 输入对应的appid 来进行杀死
3.22.通常来说,Spark与MR相比,Spark运行效率高一些,请说明效率跟高来源于Spark的内置那些机制.并列举spark运行模式?
3.23.RDD中的数据在那?
Rdd的数据存储在数据源当中,rdd只是一个抽象的数据集,我们通过RDD的操作就相当于对数据进行操作
3.24.如果对RDD进行cache操作后数据在哪里?
如果是第一次支持cache的话会被加载到各个Executor进程的内存中,第二次就会直接从内存中读取,而不是磁盘
3.25.Spark中 partition的数量有什么决定
是和Task的数量一一对应的,也可以通过参数来设置
3.26.Scala里面的函数和方法有什么区别
方法不可以作为单独的表达式存在,而函数可以,并且函数必须要有参数列表,方法可以没有参数列表,函数名只代表函数的本身,方法则是调用方法
3.27.SparkStreaming怎么进行监控?
可以通过界面进行监控,程序监控,以及API进行监控
3.28.Spark 判断shuffle的依据是什么?
是通过其父RDD的一个分区中的数据可能分配到子RDD的多个分区中
3.29.Scala有没有多继承?可以实现多继承么?
如果是class的话可以继承一个,如果是集成trait可以多继承.
3.30.SparkStreaming和Flink的区别?
Spark Streaming是基于微批次准实时处理的,所以它采用DirectDstream的方式计算出每个partition要取数据的offset范围,拉取一批数据形成RDD进行批处理.而且该RDD和kafka的分区是一一对应的
Flink是真正的流式处理,他是基于事件触发机制进行处理,在kafka Consumer拉取一批数据以后,Flink将其处理之后变成逐个Record发送的事件触发式的流处理,另外Flink支持动态的发现新增topic和partition而SparkStreaming和0.8版本的kafka结合不支持,后续和0.10版本的结合支持了
3.31. SparkContext的作用
个人理解的话,他就是Spark功能的入口,代表和spark集群的连接,能够用来在集群善上创建RDD ;累加器广播变量等
3.32.Spark Streaming读取kafka数据为什么选择直连?
因为一个RDD分区对应一个kafka分区,一个分区可以生成一个task,这个task不会丢失,会一直盯着这个分区,并且不停的读取数据
3.33.离线分析什么时候使用spark core 和spark sql?
可以根据数据来进行判断,如果你拉取的是格式化的数据可以使用sparksql非格式化的使用spark core进行计算 spark core在机器学习的迭代过程,可能会略快与spark sql 大概就是这样
3.34.Spark streaming实时数据不丢失的问题
这个问题的话,分两种情况.第一种是使用Receiver的机制保证数据的丢失,这种的是在sparkstreaming中另外启用预写日志,这样同步保存所有收到的数据到分布式的文件系统当中,以便恢复,另一种的话就是依靠Direct的方式来确保,这种是依靠checkpoint机制来确保数据不丢失,这个是在每次spark streaming消耗数据后将消费数据的offsets更新到checkpoint当中
3.35.简述宽窄依赖概念,group bykey reduce bykey,map,filter,union五种操作哪个是宽依赖,哪个是窄依赖
宽依赖:父RDD的一个分区被子RDD的多个分区使用:group by key reduce bykey 是宽依赖,会产生shuffle
窄依赖:父RDD的一个或者少量分区被子RDD的一个分区使用,例如 map filter union等操作,并不会产生shuffle
3.36.数据倾斜可能会导致那些问题,如何排查监控,设计之初要考虑那些来避免?
会导致内存溢出(OOM),导致整个任务失败.可以通过Spark的web UI来查看当前运的stage各个task分配的数量,从而进一步确定问题,亦可以通过KEY来进行统计,最开始的话可以过滤掉异常数据后进行操作,也可以提高Shuffle的并行度,这种适用于大量的key分配到同一个task上,也可以通过自定义partitioner来将key进行打散,但是这种适用与不同的key被分配到同一个Task上,也可以将Reduce端的join转化为Map端join,这样适用与某一边的数据足够小,可以被广播到各个Executor中,如果是两张表都很大的话可以使用拆分join完再union,或者将key加盐后将小表扩大N倍后join这种适用于一个数据集存在倾斜的key比较多,另外一个数据集比较平均情况下适用,
3.37.有1千万条短信,有重复,以文本的形式存储,一行一条,请用五分钟,找出重复最多的
这种的话之前了解过,我们可以采用哈希表的方法,将着1千万条分成若干组进行边扫描便建散列表,第一次扫描取首尾字节,中间随便两个字节作为hash code传入到表中并记录下地址和信息长度以及重复次数,同时Hash code且等长就是疑似相同,比较一下,相同记录只加一次进表中,但会将重复次数加一,一次扫描以后,已经记录了各自的重复次数,进行二次处理的,分组后每份中的top必须保证不同,可用hash来保证,也可直接按照hash值的大小来分类
3.38.共享变量(广播变量和累加器)
共享变量通常是向spark传递函数时,比如使用map()函数或者filter()传条件时,可以使用驱动器程序中定义变量,但是集群中运行的每个任务都会得到这些变量的一份新副本,更新这些副本也不会影响驱动器中对应变量
广播变量允许程序员将一个只读的变量缓存在每台机器上，而不用在任务之间传递变量。
累加器的话就是spark中提供的一种分布式的变量机制,其原理类似于,MR即分布式改变,然后聚合这些改变,累加器常见用途就是在调试时对作业执行过程中的事件进行计数,而广播变量就是高效分发较大的对象

3.39.当spark涉及到数据库的操作时,如何减少Spark运行中数据库连接数
可以使用for each Partition 代替foreach在foreach partition中获取数据库连接
3.40.特别大的数据怎么发送到Executor中?
3.41.Spark 优化做过那些方面的?
通常的话就是调整一下性能方面的优化,如重新分配资源,优化并行度等,Jvm调优,shuffle调优这个也是相当的重要,因为shuffle的话会占整个spark工作的50-90的百分比,在有的话就是spark调优这个的话就就是算子方面的调优,使用foreach partition代替foreach.map的话可以使用mappartition来替换,filter和coalesce的配合使用,避免数据倾斜,将reduce bykey等进行本地的聚合
3.42.Flat map底层编码实现?
底层就是将RDD每一个元素执行自定义函数f,这时这个元素的结果转换成iterator最后将这些再拼接成一个新RDD也可以理解成原本的每个元素由横向执行函数f再变为纵向,一直在回调,指到RDD内没有元素为止
3.43.Spark 1.x和2.x的区别
Data frame和dataset一体化了,只剩下Dataset了,flat map由返回list变为返回iterator了 ,foreach RDD不再返回null
更新状态的函数opation 不可再用,Spark Streaming中的ContextFactory类废弃.并且kafka的话只提供direct方式
3.44.Spark Streaming如何保证724小时运行机制
可以开启Chekpoint机制来确保724小时运行机制,这样主要是为了从失败中进行恢复,而RDDcheckpint主要是为了用状态的transformation操作,能够在其生产的数据丢失时进行快速修复
3.45.Spark Streaming是Exactloy-once么?
默认的情况并不是,而是At-least-once,但是我们可以使用幂等性以及事务来确保exactloy-once
3.46.Spark和MR区别?
Spark的话运算过程中将中间数据保存到内存当中,迭代效率略高,MR在计算过程中需要将结果落地磁盘,这样必然会有磁盘的IO操作影响效率;Spark的话容错性会高一些,因为他是弹性的数据集RDD来实现高效容错,而MR出错值能重新计算,这样成本略高,Spark和MR相比更加通用和灵活,但是Spark的生态过于复杂,比如各种RDD的差别,DAG图,stage的划分等等,而MR相比较简单,对性能要求也相对较弱.比较适合长期的后台运行,总结的话就是spark生态丰富,功能强大,性能更佳,适用范围更广,MR的话,更简单,稳定性好,适合离线的海量数据的挖掘和计算
3.47.Spark任务为何被yarn kill掉?
我之前了解到可能是某个进程占用的物理内存超过阈值然后会被yarn kill掉
3.48.Top n的具体实现步骤
这里以老师和学科举例 求出每个学科的老师topN
第一种的方法,可以先reduce bykey 进行聚合,然后在根据学科分组聚合,然后在每个分组当中进行排序求出top N
第二中的方法就是按照学科进行过滤,每次第一个学科的数据进行排序
第三种的话就是自己定义分区器,然后将相同学科的数据放到同一分区内,在对每个分区进行排序
第四中的话就是在reduce bykey的同时调用自定义分区器,这样可以减少shuffle,提高效率
3.49.怎么可以实现在一个小时Topn的固定窗口下,求出0-10 0-20?
3.50.怎么知道这个算子是转换算子还是行动算子?
转换算子的话是会返回一个新的RDD,行动算子并不会返回RDD而是直接存储数据或者是打印最终结果
3.51.有状态缓存算子?
这个的话我了解过一个状态算子比如updatesatebykey,这个的话是返回一个新的并且带有状态的Dstream
3.52.Spark什么时候用到内存?什么时候用到磁盘?
3.53.Spark的堆外内存,堆内内存
堆内内存的话

3.54.Scala尾递归?
大概了解为通过参数来传递每一次的调用结果,达到不压栈,他维护着一个迭代器和一个累加器,递归函数的返回值是递归函数表达式
4.YARN:
资源管理系统，它的基本思想是将JobTracker的两个主要功能(资源管理和作业调度/监控)分离，主要方法是创建一个全局的ResourceManager(RM)和若干个针对应用程序的ApplicationMaster(AM)
4.1.YARN 流程:
1.client向RM提交应用程序，其中包括启动该应用的AM的必须信息，例如AM程序、启动AM的命令、用户程序等。
2.RM启动一个container用于运行AM。
3.启动中的AM向RM注册自己，启动成功后与RM保持心跳。
4.AM向RM发送请求，申请相应数目的container。
5.RM返回AM的申请的containers信息。申请成功的container，由AM进行初始化。container的启动信息初始化后，AM与对应的NM通信，要求NM启动container。AM与NM保持心跳，从而对NM上运行的任务进行监控和管理。
6.container运行期间，AM对container进行监控。container通过RPC协议向对应的AM汇报自己的进度和状态等信息。
7.应用运行期间，client直接与AM通信获取应用的状态、进度更新等信息。
8.应用运行结束后，AM向RM注销自己，并允许属于它的container被收回。
4.2.YARN 调度策略:
FIFO Scheduler :不适合共享集群,大型应用会占用集群中的所有资源, 每个应用必须等待直到轮到自己运行在一个共享集群中，更适合使用容量调度器或公平调度器。
Capacity Scheduler:容量调度器允许多个组织共享一个Hadoop集群，每个组织可以分配到全部集群资源的一部分。
FairScheduler:公平调度是一种对于全局资源，对于所有应用作业来说，都均匀分配的资源分配方法