码农code之路1.讲讲你做的过的项目, 项目里有哪些难点重点呢？
kafkaDirect ES /hive kafka producer
难点值得一提的有两点：
1.rdd中用到外部变量的时候如何处理
2.广播变量的更新 rdd处理kafka读过来的数据，这些数据引用外部的class来进行规则解析，规则的更新后怎么办？
2.讲讲多线程吧， 要是你，你怎么实现一个线程池呢
Executor接口
提供了execute()方法将任务提交和任务执行分离
ExecutorService接口
继承Executor，增加了对线程池中任务生命周期的管理，可强制取消正在执行的任务，拒绝再接受任务。
提供了submit()方法来扩展Executor.execute()，使任务执行有返回值。
ScheduledExecutorService 接口
继承ExecutorService接口，在其基础上增加了定时执行任务的功能
ThreadPoolExecutor 完整的实现了一个线程池
ScheduledThreadPoolExecutor
拥有定时调度任务功能的线程池
Executors.newCachedThreadPool(); //创建一个缓冲池，缓冲池容量大小为Integer.MAX_VALUE
Executors.newSingleThreadExecutor(); //创建容量为1的缓冲池
Executors.newFixedThreadPool(int); //创建固定容量大小的缓冲池
class MyThread extends Thread {
private int i = 0;
@Override
public void run() {
for (i = 0; i < 10; i++) {
System.out.println(i);
}
}
}
class MyRunnable implements Runnable {
private int i = 0;
@Override
public void run() {
for (i = 0; i < 10; i++)
System.out.println(i);
}
}
|
3.讲一下Mapreduce或者hdfs的原理和机制。map读取数据分片。
hdfs nn dn
nn 进行元数据的管理 记录管理的文件分成了哪些块 这些块在哪些机器上（这些是DN汇报上来的）
dn 真正存储数据的节点 启动后 会向NN注册 并汇报他节点上文件block.
文件是分片的， 切割成不同的块（BLOCK 128M），每一个块默认有三个副本，副本的分布受机架感知影响，在本节点，同机架，不同机架这样分布。
客户端读取文件的时候，默认返回离其最近的一个节点来返回数据。
MR 的原理 MAP REDUCE 两种操作
map file split 的数量有关，例如文件有三个分片，而且文件本身是可分片的，就会有三个MAP 任务来处理
处理完，就会根据PARTITIONER 把数据分发到REDUCE
reduce （KEY，ITERATOR） 数据SHUFFLE过来后进行聚合处理。
4.shuffle 是什么？ 怎么调优？
shuffle operation
SPARK内特定的操作会触发SHUFFLE，如REDUCEBYKEY之类的操作。
它会触发跨执行器甚至机器之间数据的复制。
REDUCEBYKEY 会触发一个操作，对于某一个KEY的数据都要收集到特定的执行器中处理，
这就需要扫描所有的PARTITIONS。
性能影响：
因为SHUFFLE 会造成DISK／IO，数据序列化和网络IO，所以它的代价是昂贵的。
SHUFFLE会消耗大量的堆内存，因为他使用内存数据结构来组织数据，当数据大到内存承受不了
的时候，数据会被吐到磁盘上，而且增大了GC的负担。
SHUFFLE也在磁盘上生成了大量的临时文件，这些文件会被保留，一直到相对应的RDD不再被使用。
这意味着长服的作业会消耗大量的磁盘空间，这个临时存储目录是在SPARK.LOCAL.DIR.
就是Spark用于写中间数据，如RDD Cache，Shuffle，Spill等数据的位置.
Spark On YARN的时候，Spark Executor的本地路径依赖于Yarn的配置，而不取决于这个参数.
shuffle过程中常出现的问题
常见问题一：reduce oom?
问题原因：
reduce task 去map端获取数据，reduce一边拉取数据一边聚合，reduce端有一块聚合内存（executor memory * 0.2）,也就是这块内存不够
解决办法：
1.增加reduce 聚合操作的内存的比例
2.增加Executor memory的大小 –executor-memory 5G
3.减少reduce task每次拉取的数据量 设置spak.reducer.maxSizeInFlight 24m, 拉取的次数就多了，因此建立连接的次数增多，有可能会连接不上（正好赶上map task端进行GC）
常见问题二：错误描述–shuffle file cannot find or executor lost
什么时候需要调节Executor的堆外内存大小？
• shuffle file cannot find （DAGScheduler，resubmitting task）
• executor lost
• task lost
• out of memory
问题原因：
1.map task所运行的executor内存不足，导致executor 挂掉了，executor里面的BlockManager就挂掉了，导致ConnectionManager不能用，也就无法建立连接，从而不能拉取数据
2.executor并没有挂掉
2.1 BlockManage之间的连接失败（map task所运行的executor正在GC）
2.2建立连接成功，map task所运行的executor正在GC
3.reduce task向Driver中的MapOutputTracker获取shuffle file位置的时候出现了问题
解决办法：
1.增大Executor内存(即堆内内存) ，申请的堆外内存也会随之增加–executor-memory 5G
2.增大堆外内存 –conf spark.yarn.executor.memoryoverhead 2048M –conf spark.executor.memoryoverhead 2048M
(默认申请的堆外内存是Executor内存的10%，真正处理大数据的时候，这里都会出现问题，导致spark作业反复崩溃，无法运行；此时就会去调节这个参数，到至少1G（1024M），甚至说2G、4G）
如何发现数据倾斜？
有些作业异常地慢，通过Spark Web UI来查看当前运行的stage各个task分配的数据量，从而进一步确定是不是task分配的数据不均匀导致了数据倾斜。
数据倾斜解决方案
1.局部聚合和全局聚合
对热键进行多次处理，
先对热键的KEY做一个处理，加一个随机前缀，这样做一次合并，减少数据量，然后再去掉前缀再做处理的时候，数据量就会减少。
2.将 reduce join 改为 map join
A JOIB B FILTER XXX ->
例如原来使用 join 两张然后过滤数据的方式，可以使用广播变量，把过滤对象放在广播变量里面，然后在map里面做。
3. 找到具体的倾斜数据进行处理 处理方式可以参考第一种方法。
spark.shuffle.sort.bypassMergeThreshold 加大这个值，如果你的程序确实对数据不需要排序，可以加大这个参数，让其采用bypass机制，提高效率。
spark.shuffle.consolidateFiles 合并生成中间磁盘文件
spark.reducer.maxSizeInFlight 参数说明：该参数用于设置shuffle read task的buffer缓冲大小，而这个buffer缓冲决定了每次能够拉取多少数据。
关于spark shuffle 这篇文章讲解地非常不错 从原理到数据倾斜的处理方法
https://www.cnblogs.com/arachis/p/Spark_Shuffle.html
hive.map.aggr = true #Map 端部分聚合，相当于Combiner hive.groupby.skewindata=true
# 数据倾斜的时候进行负载均衡，当项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；
第二个 MR Job再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key被分布到同一个 Reduce 中），最后完成最终的聚合操作。
spark参数的优先级
在代码里面SparkConf写的优先级最高，接着是通过flags传递给spark-submit or spark-shell的，然后才是spark-defaults.conf 文件的配置。
5.项目用什么语言写？ Scala？ Scala的特点？ 和Java的区别？
关于scala，能说出以下五个特点，我觉得就对ＳＣＡＬＡ有一定的理解。
Scala /FOLDLEFT/CASE CLASS/MATCH CASE/IMPLICIT/
JVM
FUNCTIONAL PROGRAMING 函数的组合来解决问题
java 8 函数式编程的一些特性 stream() map,filter,lambda 表达式。
6.理论基础怎么样，比如数据结构里面的快速排序或者树？ 讲一讲你了解的树的知识？
快排
树
链表
7.讲一下数据库，SQl ，左外连接， 原理，实现？
有做ETL的经验，自以为对数据库还是有相对透彻的理解。擅长TSQL。
左连接：表输出的结果以左表为准，关联不到的数据，使用NULL值补充。
原理： LOOP JOIN，HASH JOIN，MERGE JOIN 在底层通过这三种方式来实现JOIN。
SQL的执行过程基本可以分为：
语法解析（PARSE) -> bind(到数据库里面去确认语法数上的元素，如表和视图,函数等对象是否存在)->优化（剪枝下移）－>执行
8.Hbase的设计有什么心得？
HBASE企业级开发指南中学习的一些技巧
ROWKEY的设计要根据查询请求来进行设计
9. 谈谈你对多线程并发问题的一些操作？
synchronized
在java中，常用这个关键字来进行锁定，便如安全的单例模式时就使用这个关键字，对NEW的方法进行锁定。
加上synchronized，并加入对象锁，new一个任意对象即可，我们这里使用Object来解决同步问题，有时候直接在方法上加的synchronized
如果这个类是一个实例类，那它使用的是什么呢？锁的是什么呢？
锁定的是this,即实例对象本身。
如果这个类或方法是静态的，那么它的原理又是什么呢？
锁定的是当前整个类本身。
有时候，也会专门new Object()来作为锁定的对象。
Java多线程可以通过：
1. synchronized关键字
2. Java.util.concurrent包中的lock接口和ReentrantLock实现类
这两种方式实现加锁。
使用synchronized 关键字有什么缺陷？
当前执行线程执行过程中，其他线程执行到同样的代码时必须等待，影响了执行效率。
粒度有点粗，譬如说要控制的数据对象有读写两个功能，就可以分读写锁，而这种方式就没有，读写操作都会被锁定。
系统自动去释放锁，第二种方法就需要用户自己适当的时候释放锁，否则就会有造成死锁的可能。
lock 可以提高多线程读的效率（或者说非排他锁的效率）
有大量线程竞争资源的时候，建议使用lock来对资源进行精细管理。
如果CLASSA有两个方法Ａ，Ｂ都加了同步的关键字，会有什么问题吗？
没有，同步关键字使用的是可重入锁，即锁是当前对象的实例，线程执行到A时获取得对象的锁，则再执行Ｂ时，就不需要再申请锁，也申请不到锁（因为前面申请的锁没有释放）。
线程Ａ来调用方法Ａ时申请了锁，还没有执行完时，同时线程Ｂ来调用方法Ｂ时能否正确执行，还是在继续等待？（等待）所以可以使用更细粒度的执行，即在方法内部对部分代码块进行锁定，锁定的时候，自己传入锁定对象，可以进一步进行控制。譬如让同时访问方法Ｂ的线程进行等待。
在方法中使用synchronized(new Object()) {} {} 的写法有什么问题吗？
经测试，还是发现它也起到了锁定的作用，线程不会同时进入方法块，
这是为什么呢？
类的静态变量是线程非安全的
类的实例变量单例是是线程非安全的，非单例时是线程安全的。
局部变量是线程安全的
尽量减少对整个方法使用synchronized关键字，我们使用代码代码块进行同步，就不影响这种情况，线程Ａ访问方法Ａ，线程Ｂ方法方法Ｂ，方法ＡＢ可都可能会有加锁的资源，但只要使用的锁对象不一样，就不影响并发操作。
11.有没有用过Zookeeper呢？ Zookeeper的适用场景是什么？
HA 状态维护 分布式锁 全局配置文件管理 操作Zookeeper是用的什么？
hbase用zookeeper存放一些配置信息 meta root信息
hadoop nn rm ha 用zk的临时节点功能和watch特性来做主备的自动切换
kafka 用zk来存放bootstrap ,customer group offset信息
12.spark开发 比如 一个读取hdfs上的文件，然后count有多少行的操作，你可以说说过程吗。那这个count是在内存中，还是磁盘中计算的呢？磁盘中。
RDD[SRING]
rdd.mappartitions(l=>l.size).collect()
13.spark sql又为什么比hive快呢？
SPARK
内存
RDD-DAG 类似于hive tez 引擎，优于ＭＲ引擎
14.RDD的数据结构是怎么样的？ Partition数组。
partitions
dependency
compute (split:Partition,taskC:TaskContext)
1.spark作业的执行过程描述一下？
作业提交之后，分析 RDD 宽窄依赖 分成不同的STAGE task 再送到EXECUTOR中去执行。
SPARK作业的提交主要涉及到driver/executor这两个节点。
driver/app master/ ->resourcemanager 交互的节点
spark app /main 运行的节点/
sparkconf->sparkcontext->server
RDD依赖分析,生成ＤＡＧ
根据DAG中的宽窄依赖，划分stage,然后生成task,分发到executor中执行。
2.Hbase的PUT的一个过程。
初始化一个HTABLE类，然后构造puT类，写入数据，第三步执行操作。
put可以插入一个或多行数据，也可以用于更新数据。
put 你怎么知道put的数据发送到哪个regionServer？
判断提交的那一个row对象是在那个region当中
RegionLocations locs = connection.locateRegion(tableName, r.getRow(), true, true, RegionReplicaUtil.DEFAULT_REPLICA_ID);
locateRegionInMeta方法到.META.表中查找tableName的row所对应的HRegion所在位置，先从本地缓存查找，如果没有，则进行下一步；
我觉得这个题最核心的部分就是这些：
行键是不可分割的字节数据，而且在表中是按由低到高的顺序排列的。
表是被切分为region分发在不同的regionServer来管理。
PUT(TABLENAME,ROWKEY,BYTESARRAY) 来找到所属REGION->RS->RPC 数据存储或放入服务端缓存。
region(startrowkey-endrowkey) tableName ,rowkey 就可台找到所属的RS
3.RDD算子里操作一个外部map比如往里面put数据。然后算子外再遍历map。有什么问题吗。
错的，关于闭包的问题说明。这个MAP就是闭包里的变量，它有自己的作用域，里面的和外面的不一样。
4.stage怎么划分的？ 宽依赖窄依赖是什么？
stage 就是根据宽窄依赖划分的，宽窄依赖是根据ＤＡＧ中任务是否产生了SHUFFLE来判断的。像filter,map,flatmap,partition ->partiton 1:1关系，不存Ｎ：Ｎ关系。
5.讲讲列式存储的 parquet文件底层格式。
parquet orc
列式存储 增大压缩比 减少数据量
可以谓词下推 提少扫描的数据量
ORC 支持ACID
PARQUET 支持嵌套的数据结构
1.scala foldLeft 方法的理解
2.scala 模式匹配
3.scala 隐式变量 隐式转换 柯里化
hbase企业应用开发实战
Hbase 安装与配置 集群运维管理 故障处理
hbase数据模型(表 行键 列族 单元格) 数据模型的操作（get put scan delete) version,sort,count,自行分区
hbase表结构设计 rowkey 列族定义 DEMO
hbase client :java client hbase shell restful client mapreduce hbase
hbase and sql :hive/hbase phoenix kundera
核心结构（b+ lsm) 过滤器 二级索引 协处理器 布隆过滤器
性能优化 JVM 查询优化 写入优化
hbase 主要有哪些配置？
1.hbase.zookeeper.quorum host:port dir
2.hbase.rootdir :数据存在hdfs哪个位置
3.是否开启集群分配式的参数 hbase.cluster.distributed
hbase 中元数据表存储tableName与region的对应关系 就像hdfs nn存储fileName与block的对应关系类似
rowkey 是不可分割的字节数组，按字典顺序由低到高存储在表中的。（很重要很重要）
有关kafka的存储？
(1).Kafka把topic中一个parition大文件分成多个小文件段，通过多个小文件段，就容易定期清除或删除已经消费完文件，减少磁盘占用。
(2).通过索引信息可以快速定位message和确定response的最大大小。
(3).通过index元数据全部映射到memory，可以避免segment file的IO磁盘操作。
(4).通过索引文件稀疏存储，可以大幅降低index文件元数据占用空间大小。
2.Kafka新建的分区会在哪个目录下创建
会在含有文件数量最少的磁盘分区创建，如果我们在log.dirs中新添加了一个目录，则新建的分区会一直往里面放，直至创建的文件数量不是最少。
partition的数据如何保存到硬盘
topic中的多个partition以文件夹的形式保存到broker，每个分区序号从0递增，且消息有序。
segment 是为了方便清理已经处理的消息。
Partition文件下有多个segment（xxx.index，xxx.log）
segment 文件里的 大小和配置文件大小一致可以根据要求修改 默认为1g
如果大小大于1g时，会滚动一个新的segment并且以上一个segment最后一条消息的偏移量命名。
索引文件里索引的是什么？
索引的是log文件的内容，具体的是有两列，第一列就是消息的序数，第二个这个消息的offset,根据索引就可以知道第几个消息在什么位置。
00000000000000170410.index 索引的名字包含了第一个全局消息offset
根据这个命名就可以通过扫描索引文件名字，找到OFFSET位于哪个索引文件，进而知道位于哪个数据文件中。
消息id由其逻辑位置决定，即从消息id可直接定位到消息的存储位置，避免id到位置的额外映射。每个part在内存中对应一个index，记录每个segment中的第一条消息偏移。
发布者发到某个topic的消息会被均匀的分布到多个part上（随机或根据用户指定的回调函数进行分布），broker收到发布消息往对应part的最后一个segment上添加该消息，
当某个segment上的消息条数达到配置值或消息发布时间超过阈值时，segment上的消息会被flush到磁盘，只有flush到磁盘上的消息订阅者才能订阅到，segment达到一定的大小后将不会再往该segment写数据，broker会创建新的segment。
kafaka生产数据时数据的分组策略
生产者决定数据产生到集群的哪个partition中，每一条消息都是以（key，value）格式，Key是由生产者发送数据传入。所以生产者（key）决定了数据产生到集群的哪个partition，如果key是空的，如何决定发送到哪个partition?
从源码可以看出，首先获取topic的所有Patition，如果客户端不指定Patition，也没有指定Key的话，
使用自增长的数字取余数的方式实现指定的Partition。这样Kafka将平均的向Partition中生产数据。
https://www.cnblogs.com/haoxinyue/p/5743775.html
消息的可靠性
当producer向leader发送数据时，可以通过request.required.acks参数来设置数据可靠性的级别：
1(默认)：这意味着producer在ISR中的leader已成功收到的数据并得到确认后发送下一条message。如果leader宕机了，则会丢失数据。
0：这意味着producer无需等待来自broker的确认而继续发送下一批消息。这种情况下数据传输效率最高，但是数据可靠性确是最低的。
-1：producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。但是这样也不能保证数据不丢失，
比如当ISR中只有leader时(前面ISR那一节讲到，ISR中的成员由于某些情况会增加也会减少，最少就只剩一个leader)，这样就变成了acks=1的情况。
关于消息传送的保障
at least once: 从broker拉取数据后，先消费，消费成功后再提交offset
at most once: 从broker拉取数据后，先提交offset，然后再消费，这样如果失败了，不会再消费第二次同样的数据
exact once: 这个比较复杂，需要配合消息的去重，建议根据业务数据的特点自行处理。
要保证数据写入到Kafka是安全的，高可靠的，需要如下的配置：
topic的配置：replication.factor>=3,即副本数至少是3个;2<=min.insync.replicas<=replication.factor
broker的配置：leader的选举条件unclean.leader.election.enable=false
producer的配置：request.required.acks=-1(all)，producer.type=sync
这篇文章讲kafka相关的内容，非常不错，内容丰富 作者可能是唯品会的员工
https://blog.csdn.net/lp284558195/article/details/80297208
hbase 面试题
hbase table 可以通过预建分区来减少合并分裂带来的性能影响，请问如何进行这项操作？
1.通过hbase shell
create ‘t1’, ‘f1′,SPLITS=>[’10’,’20’,’30’]
2.通过java api
关键问题
一是行键怎么设计，以什么样的形式生成行键。
二是行键分多少个 根据集群规则 算出会有多少个region,然后计算出区间
hbase 中的大小合并分别指的什么？ 如何在生产中调整对系统使用的影响？
hbase region server 主要做的事情就是 接收数据的读写
memstore 数据flush到磁盘后形成hfile, compaction 就是对这些文件的合并与拆分。
hbase中有两种compact
合并：
1，小合并 文件的合并
2，大合并 对某region下所有的storefile执行合并操作，这个过程中会清理有删除标记的数据和超过TTL的数据与版本的数据。
rowkey的设计原则？
1，短小 减少存储压力 无论是memostore hfile都会存储rowkey
2，散列 减少热点 尽量负载均衡 减少查询压力
3，方便业务查询 常见的需求是后插入的数据优先查询到 可以使用long.max – timestamp作为key的一部分
4，rowkey 唯一性原则 设计的时候要保证不重复
简述布隆过滤器的原理？
简单说一下BloomFilter原理
一旦表格中开启BloomFilter，那么在生成StoreFile时同时会生成一份包含BloomFilter结构的文件MetaBlock，所以会增加一定的存储开销和内存开销
内部是一个bit数组，初始值均为0
插入元素时对元素进行hash并且映射到数组中的某一个index，将其置为1，再进行多次不同的hash算法，将映射到的index置为1，同一个index只需要置1次。
查询时使用跟插入时相同的hash算法，如果在对应的index的值都为1，那么就可以认为该元素可能存在，注意，只是可能存在。
所以BlomFilter只能保证过滤掉不包含的元素，而不能保证误判包含
数据写入到的memostore,wal(log),这样才算安全写入，如果regionserver崩溃了，memostore的文件没有完成写入，可以根据wal来对数据进行重播。
那waＬ的数据写在哪里呢？一般是HDFS.
Hbase挂掉的原因很多，很多时候是因为ZK和HDFS的缘故，比如磁盘问题导致hdfs的故障。
如果rs/hm进行了长时间的gc或改动了服务器的时间，也会导致zk超时，出问题。
链接：https://www.cnblogs.com/huaxiaoyao/p/10285963.html
