Hadoop

1. MR工作原理

分为三个阶段，map,shuffle,reduce .
MR将得到的切片split分配给maptask,每个任务处理相对于的split,将split以line的方式读取每一行数据，进行mapTask运算。加载到环形缓冲区（默认为100M），双向读写（一边存元数据和索引，一边存数据），达到80M进行溢写操作期间会进行kv分区（分区数由reduce数来决定）默认使用hashpartition，再将分区中数据进行key的排序（默认排序规则是字典和升序），如果设置了setCombinerClass 则会对每个分区中的数据进行 combiner 操作，如果设置了output.compress压缩格式会对溢写的数据进行压缩，最后merge根据分区规则将数据归并到同一个文件中等待reduce的拉取。ReduceTask会启动fetch线程去Map端抓取数据，在抓取数据的时候，会只抓取当前ReduceTask所对应的分区的数据，抓取完数据之后，会对数据进行merge，将所有的数据合并到一个文件中，并且在合并过程中会进行排序 - 采用的排序机制依然是归并排序，将相同的键所对应的值放在同一组中，产生一个针对值的迭代器，这个过程称之为分组(Group)，分组完成之后，每一个键调用一次reduce方法利用reduce方法来进行处理，最后将处理结果写到HDFS上。
注意问题：
a. 缓冲区本质上是一个字节数组
b. 缓冲区是环形缓冲区，好处在于可以重复利用这个缓冲区而不用重复寻址
c.Reduce端的Shuffle
d. fetch线程数量默认为5个
f. merge因子默认为10，表示每10个文件合并成1个文件
g. ReduceTask的阈值默认为0.05，即5%的MapTask结束之后，ReduceTask就会启动开始抓取数据
(补充：切片的规则：FileInputFormat，1. 程序找到数据存储的目录 2. 遍历每一个文件 3. 获取文件大小默认情况下切片大小等于blocksize大小，blocksize在本地是32M,在集群是128M,如果文件大小大于块的1.1倍，则再开启一个切片。还有textInputFormat是按任务对文件进行规划切分，CombineInputFormat是把许多小文件放到一个切片平时默认的是TextinputFormat如果想要改变CombineInputFormat，driver中job.setInputFormatClass(CombineTextInputFormat.class)CombineTextInput.setMaxInputSplitSize(job ,20971520)把切片设置为20M)
1 客户端submit前，获取待处理的数据，根据参数配置，形成一个任务的规划
2提交信息（job.split wc.jar job.xml）给yarn中的RM
3RM计算出MapTesk的数量
4MapTesk通过默认TextInputFormat 读取reader待处理的数据，然后逻辑运算进行map过程通过Context.writer写入outputCollector给环形缓冲区
5环形缓冲区一边存索引，一边存数据，达到百分之80，分区内进行快排，然后溢写出到文件（中间可以多个combinner）
6分区内归并排序传给ReduceTask ，合并文件归并排序

Shuffle

1）Map 方法之后 Reduce 方法之前这段处理过程叫 Shuffle
2）Map 方法之后，数据首先进入到分区方法，把数据标记好分区，然后把数据发送到
环形缓冲区；环形缓冲区默认大小 100m，环形缓冲区达到 80%时，进行溢写；溢写前对数
据进行排序，排序按照对 key 的索引进行字典顺序排序，排序的手段快排；溢写产生大量溢
写文件，需要对溢写文件进行归并排序；对溢写的文件也可以进行 Combiner 操作，前提是
汇总操作，求平均值不行。最后将文件按照分区存储到磁盘，等待 Reduce 端拉取。
3）每个 Reduce 拉取 Map 端对应分区的数据。拉取数据后先存储到内存中，内存不够
了，再存储到磁盘。拉取完所有数据后，采用归并排序将内存和磁盘中的数据都进行排序。
在进入 Reduce 方法前，可以对数据进行分组操作

2.namenode,datanode,secondaryNameNode分别是干什么的？

namenode存储元数据信息，副本数，文件目录树，block 数据节点信息。
datanode数据节点，本地磁盘存储 block（文件的形式），有相关数据块的长度、效验和、时间戳，与namnode保持心跳，汇报 block 状态。secondaryNameNode协助namenode管理信息，如果做什么更改，在有个data和name的文件夹，name里边有edis和fsimage,edis里边存放的是变化操作，fsimage里边存放的是最新的元数据检查点。

3. Yarn工作原理

步骤1　用户向YARN中提交应用程序，其中包括ApplicationMaster程序、启动ApplicationMaster的命令、用户程序等。
步骤2　ResourceManager为该应用程序分配第一个Container，并与对应的Node-Manager通信，要求它在这个Container中启动应用程序的ApplicationMaster。
步骤3　ApplicationMaster首先向ResourceManager注册，这样用户可以直接通过ResourceManager查看应用程序的运行状态，然后它将为各个任务申请资源，并监控它的运行状态，直到运行结束，即重复步骤4~7。
步骤4　ApplicationMaster采用轮询的方式通过RPC协议向ResourceManager申请和领取资源。
步骤5　一旦ApplicationMaster申请到资源后，便与对应的NodeManager通信，要求它启动任务。
步骤6　NodeManager为任务设置好运行环境（包括环境变量、JAR包、二进制程序等）后，将任务启动命令写到一个脚本中，并通过运行该脚本启动任务。
步骤7　各个任务通过某个RPC协议向ApplicationMaster汇报自己的状态和进度，以让ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。
在应用程序运行过程中，用户可随时通过RPC向ApplicationMaster查询应用程序的当前运行状态。
步骤8　应用程序运行完成后，ApplicationMaster向ResourceManager注销并关闭自己。

4. hdfs的工作流程

读流程：

client创建一个代理对象与namenode进行rpc通信，拿Nn的源信息，Nn校验无误后，给，然后client获取元信息对应的block块，最后将block合并成一个文件进行返回。
1 客户端创建一个代理对象 DistributedFileSystem
2 通过rpc连接namenode namenode校验完毕之后，返回给客户端。
3客户端获取元数据信息后，请求读取block1 ,block传输数据到 FSdatainputformatStream 继续请求读取block2，最终合并成一个文件进行返回。
（RPC和HTTP两种传输协议，RPC主要基于TCP/UDP协议；
HTTP协议是应用层协议，是构建在传输层协议TCP之上的；
从效率来看的话RPC更胜一筹！RPC长连接：不必每次通信都像http一样去三次握手，减少网络开销；HTTP服务开发迭代更快：在接口不多，系统与系统之间交互比较少的情况下，http就显得更加方便；相反，在接口比较多，系统与系统之间交互比较多的情况下，http就没有RPC有优势。）
写流程：client与namenode进行rpc通信，在namenode上请求创建元信息，namenode触发副本放置策略，返回元数据信息，client和datanode链接，client把packet放入到一个队列中，并向第一个datanode发送packet,这一个过程中上有节点同时发下一个packet,当block传输完成时，dn向各自的nn汇报，同时client持续传输下一个,所以client和block的汇报也是并行的。

写流程

1 client 创建一个代理对象,rpc请求与namenode的连接 2 namenode验证后响应可以上传文件。 3 client上传第一个block块 4 namenode 返回几个节点存储数据

5 client创建FSdataoutputformatstream 与datanode请求连接，datanode响应，开始传输packet给第一个块，第一个块传给后面的。

5.block块副本放置策略

第一个副本：放置在上传文件的 DataNode；如果时集群外提交，则随机挑选一  台磁盘不太满，CPU 不太忙的节点。
第二个副本：放置在于第一个副本不同的机架的节点上。
第三个副本：与第二个副本相同机架的节点。

combiner和partition的作用：

combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量
partition的默认实现是hashpartition，是map端将数据按照reduce个数取余，进行分区，不同的reduce来copy自己的数据。
partition的作用是将数据分到不同的reduce进行计算，加快计算效果。

namenode的safemode是怎么回事？如何才能退出safemode？

namenode在刚启动的时候元数据只有文件块信息，没有文件所在datanode的信息，需要datanode自己向namenode汇报。如果namenode发现datanode汇报的文件块信息没有达到namenode内存中所有文件块的总阈值的一个百分比，namenode就会处于safemode。
只有达到这个阈值，namenode才会推出safemode。也可手动强制退出。

一个datanode 宕机,怎么一个流程恢复？

Datanode宕机了后，如果是短暂的宕机，可以实现写好脚本监控，将它启动起来。如果是长时间宕机了，那么datanode上的数据应该已经被备份到其他机器了，
那这台datanode就是一台新的datanode了，删除他的所有数据文件和状态文件，重新启动

Hadoop宕机

1如果MR造成系统宕机，此时要控制yarn同时运行的任务数目，和每个任务申请的最大内存，调整参数yarn.scheduler.maximum-allocation-mb（单个任务可申请的最多物理内存量，默认是 8192MB）
2如果写入文件过量造成NameNode宕机，那么调高Kafka的存储大小，控制从Kafka到HDFS的写入速度，高峰时用Kafka进行缓存。

hadoop 的 namenode 宕机,怎么解决？

先分析宕机后的损失，宕机后直接导致client无法访问，内存中的元数据丢失，但是硬盘中的元数据应该还存在，如果只是节点挂了，
重启即可，如果是机器挂了，重启机器后看节点是否能重启，不能重启就要找到原因修复了。
但是最终的解决方案应该是在设计集群的初期就考虑到这个问题，做namenode的HA。

Hadoop数据倾斜问题？

1）提前在 map 进行 combine，减少传输的数据量
在 Mapper 加上 combiner 相当于提前进行 reduce，即把一个 Mapper 中的相同 key 进行
了聚合，减少 shuffle 过程中传输的数据量，以及 Reducer 端的计算量。
如果导致数据倾斜的 key 大量分布在不同的 mapper 的时候，这种方法就不是很有效了。
2）导致数据倾斜的 key 大量分布在不同的 mapper
（1）局部聚合加全局聚合。
第一次在 map 阶段对那些导致了数据倾斜的 key 加上 1 到 n 的随机前缀，这样本来相同的 key 也会被分到多个 Reducer 中进行局部聚合，数量就会大大降低。
第二次 mapreduce，去掉 key 的随机前缀，进行全局聚合。
思想：二次 mr，第一次将 key 随机散列到不同 reducer 进行处理达到负载均衡目的。第二次再根据去掉 key 的随机前缀，按原 key 进行 reduce 处理。
这个方法进行两次 mapreduce，性能稍差。
（2）增加 Reducer，提升并行度
JobConf.setNumReduceTasks(int)
（3）实现自定义分区
尚硅谷大数据技术之高频面试题
根据数据分布情况，自定义散列函数，将 key 均匀分配到不同 Reducer

Hadoop常用端口号

dfs.namenode.http-address:50070 9870
8020 9820 运行hdfs时访问namenode时，也就是内部端口号
dfs.namenode.http-address:50075 ：9864
dfs.namenode.address:50010:9866
SecondaryName 辅助名称节点端口号 50090 9868
yarn.resourcemanager.webapp.address:8088
历史服务器 web 访问端口：19888

hadoop配置文件

配置文件：yarn-site.xml（开启日志聚集，日志保留时长，namenode获取数据方式，RM的地址） hdfs-site.xml （副本数）core-site.xml（临时文件tmp,namenode的端口号） mapred-site.xml（将NM交给yarn,历史服务器地址，历史服务器web地址）
脚本文件：hadoop-env.sh（export JVM地址） yarn-env.sh mapred-env.sh
集群名称：slaves

集群搭建流程

配置jvm
ssh免密
配置hadoop
hdfs format namenode

yarn的默认调度器，调度器分类，以及他们之间的区别

1FIFO
2Capacity Schedule（容量调度器）Hadoop2.7.2中默认的是这个，多个队列并行运行
3Fair Scedule（公平调度器）和容量调度器不同的是，他是按照缺额对每个任务先后分配资源的。（缺额=理论上应该有的资源减去实际上分配的资源）

Hadoop的优化

面对许多小文件

HDFS小文件的影响
（1）默认的是TextInputFormat，会开启许多map
(2) 影响namenode的寿命，因为文件元数据存储在Map任务中。
解决办法：
数据输入文件处理：
1 JVM重用
（JVM重用，解决小文件过多问题，每一个task都会启动一个JVM,当一个job中有过多的mapTask中，会启动较多的JVM,这时候JVM刚启动，就结束了，jvm每次启动大概需要1s的时间，为了不让它开开停停，所以它要配置重用来提高效率，在maper-site.xml中配置，如果task属于不同的job，那么JVM重用机制无效，不同job的task需要不同的JVM来运行。）
2 可以用CombinFileInputFormat来获取数据
3 合并小文件，自定义Inputformat将小文件存储成SequecnceFile文件
Map阶段：
1 增大环形缓冲区大小，由100m扩大到200m
2增大环形缓冲区溢写比例 80%到90%
3减少对溢写文件的merge次数
4不影响实际业务的前提下，采用Combiner提前合并，减少I/o
Reduce阶段
1合理设置Map和Reduce数：两个都不能设置太少，也不能设置太多
2设置Map,Reduce共存调整slowstart.completemaps参数。
3规避使用reduce,因为Reduce在连接数据的时候会产生大量的网络消耗
4集群性能可以的前提下，增大Reduce端存储数据内存的大小。
IO传输
1采用数据压缩的方式，减少网络IO的时间，安装Snappy和LZOP压缩编码机器
2使用SequenceFile二进制文件
整体
1MapTesk默认内存大小为1G,可以增加MapTask内存大小为4-5g
2ReduceTask默认内存大小为1G,可以增加ReduceTask内存大小为4-5g
3增加MapTask的cpu核数，增加ReduceTask的CPU数
4增加每个container的CPU核数和内存大小。
5调整每个MapTask和ReduceTask最大重视次数
记忆压缩：常见的由Gzip 和Bzip2 然后deflate也是默认的，LZO需要建立索引，需制定输入格式(只有LZO和Bzip能切分），Snappy是个快乐的小鳄鱼，速度块，缺点无法分割。

Hadoop参数调优

1）在 hdfs-site.xml 文件中配置多目录，最好提前配置好，否则更改目录需要重新启动
集群
2）NameNode 有一个工作线程池，用来处理不同 DataNode 的并发心跳以及客户端并发
的元数据操作。
dfs.namenode.handler.count=20 * log2(Cluster Size)，比如集群规模为 10 台时，此参数设
置为 60
3 ）编辑日志存储路径 dfs.namenode.edits.dir 设置与镜像文件存储路径
dfs.namenode.name.dir 尽量分开，达到最低写入延迟
4）服务器节点上 YARN 可使用的物理内存总量，默认是 8192（MB），注意，如果你
的节点内存资源不够 8GB，则需要调减小这个值，而 YARN 不会智能的探测节点的物理内存总量。yarn.nodemanager.resource.memory-mb
5）单个任务可申请的最多物理内存量，默认是 8192（MB）。yarn.scheduler.maximumallocation-mb

Spark常见的面试题

RDD是什么？创建方式？分别怎么分区的？

rdd是一个弹性分布式数据集，不可变可分区，内部元素可并行计算的集合。makeRDD，parallelize 或者由上一个rdd得出的结果创建，
分区方式1.由本地核数分区，2使用外部hdfs数据源，由block块决定，默认两个分区，由另一个rdd得出的结果创建，既转换时创建，根据父rdd的reduceTask数量创建。分区数默认为2 ，如果比2小，则采用更小的，如果比2大就进行下面的运算。实际上，底层用的是hadoop,的Textinputformat 读取数据,按照任务进行分区
4个文件长度分别为100 100 100 1400字节,默认最小分区为2
首先计算全部文件总长度totalSize=100+100+100+1400=1700
goalSize=totalSize/最小分区数即2 =850
blockSize=128M换算成字节为134217728
minSize=1
goalSize与blockSize取最小值为850
850 与minSize取最大值为850
即splitSize为850
然后每个文件长度除以850 判断是否大于1.1
文件1,2,3都是100所以各生成1个分区,
文件4位1400,除以850>1.1 切分一个分区,剩余
(1400-850)/850 >1.1不再成立又生成一个分区.
所以举例中的四个文件共生成5个分区

2. task stage job分别是什么？

即 stage 下的一个任务执行单元，一般来说，一个 rdd 有多少个 partition，就会有多少个 task，因为每一个 task 只是处理一个partition 上的数据。一个Job会被拆分为多组Task，每组任务被称为一个stage，每一次数据的shuffle都会产生一个stage。每触发一次action操作就会生成一个job，
为什么要划分stage?
由于划分完 stage 之后，在同一个stage中只有窄依赖，没有宽依赖，可以实现流水线计算，stage中的每一个分区对应一个task，在同一个stage中就有很多可以并行运行的task。

3. RDD的缓存持久机制

cache persist checkpoint
cache是基于内存的persist
chche () = persist()=persist(StorageLevel.Memory_Only)
persist可以指定持久化的级别。最常用的是MEMORY_ONLY和MEMORY_AND_DISK。”_2”表示有副本数

4. spark常见调优

1 修改序列化机制有效压缩数据量，通过kryo优化序列调优
conf.set(“spark.serializer”,"org.apache.spark.serializer.KryoSerializer ")
2 提交spark时，在里边调整对应的参数

在这里插入代码片  spark-submit \--master spark://node1:7077 \--class com.kaikeba.WordCount \--num-executors 3 \    配置executor的数量--driver-memory 1g \   配置driver的内存（影响不大）--executor-memory 1g \ 配置每一个executor的内存大小--executor-cores 3 \   配置每一个executor的cpu个数/export/servers/wordcount.jar

3 提高并行度
1 设置tasks数量
new SparkConf().set(“spark.defalut.parallelism”,“500”)

2 给RDD重新设置新的分区repartition(增大分区数目)增大task，提高并行度
3 通过设置参数sql.shuffle.partitions

4RDD的重用和持久化
就是使用persisit,和cache,checkpoint
5使用广播变量
比如一个任务需要50个executor，1000个task，共享数据为100M。

在不使用广播变量的情况下，1000个task，就需要该共享数据的1000个副本， 也就是说有1000份数需要大量的网络传输和内存开销存储。耗费的内存大小 1000*100=100G.
使用了广播变量后，50个executor就只需要50个副本数据，而且不一定都是 从Driver传输到每个节点，还可能是就近从最近的节点的executor的 blockmanager上拉取广播变量副本，网络传输速度大大增加；内存开销 50*100M

注意事项：能不能将一个RDD使用广播变量广播出去？不能，因为RDD是不存储数据的。可以将RDD的结果广播出去。广播变量只能在Driver端定义，不能在Executor端定义。在Driver端可以修改广播变量的值，在Executor端无法修改广播变量的值。如果executor端用到了Driver的变量，如果不使用广播变量在Executor 有多少task就有多少Driver端的变量副本。如果Executor端用到了Driver的变量，如果使用广播变量在每个Executor中只有一份Driver端的变量副本。

6shuffle调优（–此处复杂，以后再说）
7使用高性能算子
使用reduceByKey/aggregateByKey替代groupByKey，使用mapPartitions替代普通map ,使用filter之后进行coalesce操作从而减少分区数目

HIve

hive内部表和外部表的区别

内部表：加载数据到hive所在的hdfs目录，删除时，元数据和数据文件都删除
外部表：不加载数据到hive所在的hdfs目录，删除时，只删除表结构。

分桶的作用

最大的作用是提高join的效率。（1）获得更高的查询处理效率。（2）使取样（sampling）更高效。

Hive 你们用的是外部表还是内部表,有没有写过UDF。

UDF：
1、写对应的java代码自定义函数的逻辑
2、将代码打成jar包上传到hive
3、在hive创建临时函数与对应的class类相关联
4、在hive中调用临时函数。

KAFKA

1.Kafka中的ISR(InSyncRepli)、OSR(OutSyncRepli)、AR(AllRepli)代表什么？

ISR 速率和leader相差低于10s的follower的集合
OSR 速率和leader相差大于10s的follwer集合
AR 所有分区的follower

Kafka种的HW,LEO等分别代表什么？

HW 又名高水位，根据同一分区中，最低的LEO决定
LEO ：每个分区的最高offset.

Kafka的用途有哪些？使用场景如何？

1用户追踪，根据用户在web或者app上的操作，将这些操作消息记录到各个topic中，然后消费者通过订阅做数据挖掘
2日志收集
3消息系统：缓存消息
4运营指标，记录运营监控数据，如报错的报告；

kafka怎样体现消息的顺序性的？

每个分区内，每条消息都有offset,所以只能在同一分区内有序，但不同分区无法做到消息的顺序性。

消费组中的消费者个数如果超过topic的分区，那么就会有消费者消费不到数据这句话是否正确？

对超过的分区数的消费者不会再接收到数据

有哪些情形会造成重复消费，或丢失信息？

先处理后提交offset,会造成重度消费
先提交offset后处理，会造成数据丢失

KAFKA分区的目的

对于kafka集群来说，分区可以做到负载均衡，对于消费者来说，可以提高并行度，提高读取效率。

KAFKA的高可靠性是怎么实现的？

为了实现高可靠性，kafka使用了订阅的模式，并使用了isr和ack应答机制
能进入isr中的follower和leader之间的速率不会相差10s.
当ack=0时，producer不等待broker的ack,不管数据有没有写入成功，都不再重复发该数据
当ack=1时，broker会等到leader写完数据后，就会向producer发送ack,但不会等follower同步数据，如果这时leader挂掉，producer会对新的leader发送新的数据，在old的leader中不同步的数据就会丢失。
当ack=-1，或者all时，broker会等到leader和isr中的所有follower都同步完数据，再向producer发送ack,有可能造成数据重复。

topic的分区数目可不可以增加？

bin/kafka-topics.sh --zookeeper localhost:2181/kafka --alter --topic topic-config --partitions 3

topic分区数可不可以减少？

不可以，先有的分区数据难以处理。

简述Kafka的日志目录结构

每一个分区对应一个文件夹，命名为topic-0,topic-1，每个文件夹有.index 和.log文件

如何解决消费者速率低的问题

增加分区数和消费者数

kafka有哪些设计让他有如此高的性能

1kafka是分布式的消息队列
2 对log文件进行了segment，并对segment建立了索引。
3（对于单节点）使用顺序读写，速度可以达到600M/s
4 引动了zero拷贝，在os系统就完成了读写操作

kafka启动不起来的原因

在关闭kafka时，先关了zookeeper，就会导致下次启动时，会报节点已存在的错误，只需把zookeeper中的zkdata/version-2的文件夹删除即可。

kafkaController的作用

负责kafka集群的上下线工作，所有的topic的副本分区分配和选举leader的工作

失效副本是什么？有哪些应对的措施？

失效副本为速率比leader相差大10s的follower
将失效的follower先提出isr
等速率接近leader10s内，再加进isr.

Kafka消息采用的是Pull模式，还是Push模式？

在producer阶段，是向broker用push
consumer 阶段是向broker用pull
在Pull模式下，consumer可以根据自身速率选择拉取数据，避免了低速率的consumer发生崩溃的问题
但缺点是,consumer要时不时的去询问broker是否有新数据,容易发生死循环,内存溢出

Kafka创建Topic时如何将分区放置到不同的broker中?

首先副本数不能超过broker数
第一分区是随机从broker中选择一个，然后其他分区相对于0号分区依次向后移第一个分区是从nextReplicaShift决定的，而这个数也是随机产生的。

kafka中的事务是怎么实现的

kafka 事务有两种
producer事务和consumer事务
producer事务是为了解决kafka跨分区跨会话问题
kafka不能跨分区跨会话的主要问题是每次启动的producer的PID都是系统随机给的
所以为了解决
手动给producer一个全局唯一的id,也就是transcation id 简称TID 我们将TID和PID进行绑定，在producer带着PID和TID第一次向broker注册时，broker就会记录TID,并生成一个新的组件_transaction_state用来保存TID的事务状态信息
当producer重启后，就会带着TID和新的PID向broker发起请求，当发现Tid一致时，producer就会获取之前的PID,并覆盖。
consumer事务相对于producer事务就弱一点，需要先确保consumer的消费和提交位置为一致且具有事务功能，才能保证数据的完整性，不然会造成数据的丢失和重复

Kafka中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？

拦截器>序列化器>分区器

Kafka生产者客户端的整体结构是什么样子的？使用了几个线程来处理？分别是什么？

main线程和sender线程
main线程会依次经过拦截器,序列化器,分区器将数据发送到RecourdAccumlator(线程共享变量)
再由sender线程从RecourdAccumlator中拉取数据发送到kafka broker
相关参数：
batch.size：只有数据积累到batch.size之后，sender才会发送数据。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。

消费者提交消费位移时提交的是当前消费到的最新消息的offset还是offset+1？

offset+1

Kafka 的机器数量

Kafka 机器数量=2*（峰值生产速度*副本数/100）+1

Kafka压测

Kafka 官方自带压力测试脚本（kafka-consumer-perf-test.sh、kafka-producer-perf-test.sh）。
Kafka 压测时，可以查看到哪个地方出现了瓶颈（CPU，内存，网络 IO）。一般都是网络 IO
达到瓶颈

Kafka 的日志保存时间

默认7天

Kafka的硬盘大小

每天的数据量*7/70%

Kafka监控

公司自己开发的监控器；开源的监控器：KafkaManager、KafkaMonitor、kafkaeag

Kafka分区分配策略

在kafka内部两种默认的分区分配策略
Range RoundRobin
Range是默认的Range 是默认策略。Range 是对每个 Topic 而言的（即一个 Topic 一个 Topic 分），首先
对同一个 Topic 里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。然后用
Partitions 分区的个数除以消费者线程的总数来决定每个消费者线程消费几个分区。如果除
不尽，那么前面几个消费者线程将会多消费一个分区。
例如：我们有 10 个分区，两个消费者（C1，C2），3 个消费者线程，10 / 3 = 3 而且除
不尽。
C1-0 将消费 0, 1, 2, 3 分区
C2-0 将消费 4, 5, 6 分区
C2-1 将消费 7, 8, 9 分区
第一步：将所有主题分区组成 TopicAndPartition 列表，然后对 TopicAndPartition 列表按
照 hashCode 进行排序，最后按照轮询的方式发给每一个消费线程。

Kafka 消息数据积压，Kafka 消费能力不足怎么处理？

1）如果是 Kafka 消费能力不足，则可以考虑增加 Topic 的分区数，并且同时提升消费
组的消费者数量，消费者数=分区数。（两者缺一不可）
2）如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取
数据/处理时间<生产速度），使处理的数据小于生产的数据，也会造成数据积压

Kafka的幂等性

Producer 的幂等性指的是当发送同一条消息时，数据在 Server 端只会被持久化一次，数
据不丟不重，但是这里的幂等性是有条件的
1）只能保证 Producer 在单个会话内不丟不重，如果 Producer 出现意外挂掉再重启是
无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不
丢不重）。
2）幂等性不能跨多个 Topic-Partition，只能保证单个 Partition 内的幂等性，当涉及多个
Topic-Partition 时，这中间的状态并没有同步。

Kafka事务

Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在 Exactly Once 语义的基
础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败。
1）Producer 事务
为了实现跨分区跨会话的事务，需要引入一个全局唯一的 Transaction ID，并将 Producer
获得的PID 和Transaction ID 绑定。这样当 Producer重启后就可以通过正在进行的 Transaction
ID 获得原来的 PID。
为了管理 Transaction，Kafka 引入了一个新的组件 Transaction Coordinator。Producer 就
是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。Transaction
Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic，这样即使整个服务重启，由于
事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。
2）Consumer 事务
上述事务机制主要是从 Producer 方面考虑，对于 Consumer 而言，事务的保证就会相对
较弱，尤其时无法保证 Commit 的信息被精确消费。这是由于 Consumer 可以通过 offset 访
问任意信息，而且不同的 Segment File 生命周期不同，同一事务的消息可能会出现重启后被
删除的情况。

Kafka高效读写数据

1）Kafka 本身是分布式集群，同时采用分区技术，并发度高。
2）顺序写磁盘
Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，
为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这
与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。
3）零复制技术

Kafka 挂掉

1）Flume 记录
2）日志有记录
3）短期没事

Linux&Shell

linux常用命令

top 查看内存
df -h 查看磁盘存储情况
iotop 查看磁盘IO读写（yum install iotop安装）
iotop -o 直接查看比较高的磁盘读写程序
netstat -tunlp | grep 端口号查看端口占用情况
uptime 查看报告系统运行时长及平均负载
ps aux 查看进程

shell常用命令

awk，sed,cut,sort（具体请看我的主页）

Flume

1Flume组成，Put事务，Take事务

Taildir Source ：断点续传，多目录。Flume1.6以前需要自己定义source记录每次读取文件位置，实现断点续传。
FileChannel ：数据存储在磁盘，宕机数据可以保存，但是传输速率慢。适合对数据传输可靠性要求高的场景，比如，金融行业。
MemoryChannel
KafkaChannel:减少了Flume的Sink阶段，提高了传输效率。
Source到Channel是put事务
Channel到sink是take事务

2Flume拦截器

1拦截器注意事项
项目中自定义了：ETL拦截器和区分类型拦截器
采用两个拦截器的优缺点：优点，模块化开发和可移植性；缺点，性能会低一些
2自定义拦截器步骤
a实现Interceptor
b 重写四个方法
initialize初始化
public Event intercept(Event event)处理单个Event
public List intercept(List events)处理多个Event,在这个方法中，调用Event intercept(Event event)
close方法
c静态内部类实现Interceptor Builder

3FlumeChannel拦截器

ChannelSelectors 可以让不同项目日志通过不同的Channel到不同的Sink中去，官方上Channel Selectors 有两种类型：Replicating Channel Selector(default)和Multiplexing Channel Selector
这两种Selector的区别是:Replicating 会将source过来的events发往所有channel,而Multiplexing可以选择该发往哪些Channel。

4Flume监控器

Ganglia

5Flume采集会数据丢失么?

不会，Channel存储可以存储在File中，数据传输自身有事务。

6Flume内存

开发中在flume-env.sh中设置JVM heap为4G或更高，部署在单独的服务器上（4核8线程16G内存）
-Xms与-Xmx最好设置一致，减少内存抖动带来的性能影响，如果设置不一致容易导致频繁fullgc

7FileChannel优化

通过配置dataDirs指向多个路径，每个路径对应不同的硬盘，增大Flume吞吐量。checkpointDir 和 backupCheckpointDir 也尽量配置在不同硬盘对应的目录中，保证
checkpoint 坏掉后，可以快速使用 backupCheckpointDir 恢复数据’

位图和布隆

位图是用来保存状态，去重排序
缺点1数组下标有限制 URL,name
2 一定是随着最大数据的变化而变化
布隆过滤器，简单来说就是把URL通过hash转换为数组下标，保存在位图中。
会碰到hash碰撞，布隆过滤器是怎么避免的呢？布隆过滤器除了一个位数组，还有K个hash函数，当一个元素加入布隆过滤器中的时候，会进行如下操作：
-使用k个hash函数对元素值进行K次计算，得到K个哈希值
-根据得到的hash值，在位数组中把对应的下标值设置为1.

private static int size=100000000  //预计要插入多少数据
private static double fpp =0.01 //误判的频率
private static BloomFilter<Integer> bloomFilter =BloomFilter.create(Funnels.integerFunnel(),size,fpp);
public static void main(String[] args){//插入数据
for(int i=0; i<100000;i++){bloomfilter.put(i);
}
bloomFilter.mightContain(100)
}

线性回归算法

有监督的算法 ：指的是训练集中含有y值，比如说线性回归

先理解什么是回归就是通过结果求参数
得到y=w0+w1x 然后error=前m项目和(y-h(x))^2最小，则数值最准。
怎么调节呢 w=w1-步长*error的导数步长不能太大，不能太小

import org.apache.log4j.Logger
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.event.Level
import org.apache.spark.mllib.regression.{LabeledPoint, LinearRegressionModel, LinearRegressionWithSGD}
object ss {def main(args: Array[String]): Unit = {//构建sparkval conf=new SparkConf().setAppName("").setMaster("local[*]")val sc=new SparkContext(conf)sc.setLogLevel("ERROR")//读取样本数据val data_path="E:\\work\\sm\\src\\main\\resources\\lpsa.data"val data=sc.textFile(data_path)val examples=data.map{line=>val parts=line.split(",")val y=parts(0)val xs=parts(1)//Vectors矢量 有两种 Vectors.sparse()稀疏向量 ，dense稠密向量//LabledPoint标记点，第一个参数是 label,第二个参数是feather 一般在监督学习和回归中用LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(_.toDouble)))}val train2TestData=examples.randomSplit(Array(0.8,0.2),1L)/*迭代次数训练一个多元线性回归模型收敛（停止迭代）条件1.error的值小于用户指定的error值2.达到一定的迭代次数*/val numItercept=1000 //最大迭代次数//这个参数就是当80%训练集仍旧很大时，选择百分之多少进行error计算val miniBathFraction=1val stepSize=0.0001//BGD和SGD,SGD样本随机抽取一部分计算val lrs=new LinearRegressionWithSGD()lrs.setIntercept(true) //设置截距提高精确度lrs.optimizer.setStepSize(stepSize)lrs.optimizer.setNumIterations(numItercept)//Batch 批量 Fraction 分数lrs.optimizer.setMiniBatchFraction(miniBathFraction)train2TestData(0).foreach(print)val model =lrs.run(train2TestData(0))println("weights="+ model.weights)println("intercept"+model.intercept)//对样本开始测试val prediction =model.predict(train2TestData(1).map(_.features))val predictionAndLabel=prediction.zip(train2TestData(1).map(_.label))val print_predict =predictionAndLabel.take(20)for(i <- 0 to print_predict.length - 1){println(print_predict(i)._1 + "\t" + print_predict(i)._2)}//计算平均误差val loss=predictionAndLabel.map{case(p,v)=>val err = p - vMath.abs(err)}.reduce(_+_)val error=loss / train2TestData(1).countprintln("error" + error)sc.stop()}}

贝叶斯

朴素贝叶斯（Naive Bayes ,NB）算法是基于贝叶斯定理与特征条件独立假设的分类方法，该算法是有监督的学习算法，解决的是分类问题，是将一个未知样本分到几个预先已知类别的过程。
朴素贝叶斯的思想就是根据某些个先验概率计算Y变量属于某个类别的后验概率，也就是根据先前事件的有关数据估计未来某个事件发生的概率。

原理：哪个概率高，就分到哪一类

拉普拉斯估计
拉普拉斯估计本质上是给频率表中的每个单词的计数加上一个较小的数，这样就保证每一类中每个特征发生的概率非零。通常，拉普拉斯估计中加上的数值为1，这样就保证了每一个特征至少在数据中出现一次。

KNN 最近邻算法

NearestNeighbors

1.KNN算法步骤：

1)对于未知类别的数据（对象，点），计算已知类别数据集中的点到该点的距离。
2)按照距离由小到大排序
3)选取与当前点距离最小的K个点
4)确定前K个点所在类别出现的概率
5)返回当前K个点出现频率最高的类别作为当前点预测分类
3.KNN算法复杂度：
KNN 分类的计算复杂度和训练集中的文档数目成正比，也就是说，如果训练集中文档总数为 n，那么 KNN 的分类时间复杂度为O(n)
4.KNN问题：
该算法在分类时有个主要的不足是，当样本不平衡时，如一个类的样本容量很大，而其他类样本容量很小时，有可能导致当输入一个新样本时，该样本的 K 个邻居中大容量类的样本占多数。解决：可以采用权值的方法，根据和该样本距离的远近，对近邻进行加权，距离越小的邻居权值大，权重一般为距离平方的倒数。
5.KNN数据归一化：
为了防止某一维度的数据的数值大小对距离计算产生影响，保证多个维度的特征是等权重的，最终结果不能被数值的大小影响，应该将各个维度进行数据的归一化，把数据归一化到[0,1]区间上。
归一化公
式：

K-means

算法思想：
以空间中K个点为中心进行聚类，对最靠近他们的对象归类，通过迭代的方法，逐次更新各聚类中心的值，直到得到最好的聚类结果
算法流程总结：
1、适当选择k个类的初始中心
2、对任意一个样本，求其到各类中心的距离，将该样本归到距离最短的中心所在的类
3、利用均值等方法更新每个聚类的中心值
4、重复2,3的迭代，直到k个中心点值保持不变，则迭代结束，否则继续迭代

Kmeans++算法的介绍

从输入的数据点集合中随机选择一个点作为第一个聚类中心
对于数据集中的每一个点x，计算它与最近聚类中心(指已选择的聚类中心)的距离D(x)
选择一个新的数据点作为新的聚类中心，选择的原则是：D(x)较大的点，被选取作为聚类中心的概率较大
重复2和3直到k个聚类中心被选出来
利用这k个初始的聚类中心来运行标准的k-means算法

TF-IDF

逻辑算法

二战

RPC

RPC主要是传输数据 stub 和runtime stub封装了底层的细节，包括参数的序列化和网络传输的一些过程，runtime主要是传输过程中的一些网络调用。
适用场景：内部系统中一些同步调用
内部系统中的一些异步用kafka mq
外部系统 restful 在这里插入图片描述

Hadoop

1 hdfs写流程
client向namenode通过rpc发送请求，namenode根据路径和是否存在目录判断是否可以写。client将文件以128M拆分，然后向namenode请求datanode的地址，namenode给地址，（本地一份，相同机架一份，不用机架一份）client通过rpc进行传输，每个packge(64kb)传给本地，本地再给其他两个备份。直到第一个blocK传输完毕，开始进行第二个传输。
2 hdfs读流程

java

hashmap

大数据面试题（每天更新，直到我找到工作）相关推荐

大数据面试题及答案 100道（2021最新版）
大数据面试题及答案[最新版]大数据高级面试题大全(2021版),发现网上很多大数据面试题都没有答案,所以花了很长时间搜集,本套大数据面试题大全最近由于要准备面试就开始提早看些面试.笔试题.以下是自己 ...
大数据面试题及答案汇总版
版权声明:本文为博主原创文章,遵循 CC 4.0 by-sa 版权协议,转载请附上原文出处链接和本声明. 本文链接:https://blog.csdn.net/albg_boy/article/det ...
rdd分片 spark_大数据面试题（Spark（一））
大数据面试题(Spark(一)) 大家好,我是蓦然,这一系列大数据面试题是我秋招时自己总结准备的,后续我会总结出PDF版,希望对大家有帮助!1.spark的有几种部署模式,每种模式特点?(☆☆☆☆☆) ...
数据分析大数据面试题大杂烩01
互联网:通过埋点实时计算用户浏览频次用优惠券等措施吸引用户,通过历史信息用非智能学习的title方式构造用户画像(抖音,京东) 电信,银行统计营收和针对用户的个人画像:处理大量非实时数据政府:健康码 ...
大数据面试题_Hive篇
文章目录一.大数据面试题_Hive篇 (一)Hive表与性能优化 1.hive 内部表和外部表的区别 2. hive 有索引吗 3.sort by 和 order by 的区别 4.如何使用过 Hi ...
5W字讲解最新大数据面试题
此套面试题来自于各大厂的真实面试题及常问的知识点,如果能理解吃透这些问题,你的大数据能力将会大大提升,进入大厂指日可待复习大数据面试题,看这一套就够了! 本文目录: 一.Hadoop 二.Hive ...
2021最全大数据面试题汇总---hadoop篇,附答案！
大数据面试题来了! 本篇文章搜集了常见的大数据面试题以及答案,包含了Hadoop,Flume,Hbase,Hive,kafka,spark,zookeeper等方面的内容,助同学们收到心仪的Offer ...
常见大数据面试题汇总带答案
大数据面试题汇总 **牛客网刷sql题** redis flume flink(大部分知识点写过的帖子里都有) mr,java ,集群算法题 kafka 维度建模分为哪几种? SQL高频面试题 hb ...
深圳公司大数据面试题汇总
点击上方 "大数据肌肉猿"关注, 星标一起成长后台回复[加群],进入高质量学习交流群 2021年大数据肌肉猿公众号奖励制度相关阅读: 上海小公司大数据面试题汇总背景本文为# ...
大数据知识面试题-MapReduce和YARN（2022版）
序列号内容链接 1 大数据知识面试题-通用(2022版) https://blog.csdn.net/qq_43061290/article/details/124819089 2 大数据知识面试 ...

大数据面试题（每天更新，直到我找到工作）

Hadoop

1. MR工作原理

Shuffle

2.namenode,datanode,secondaryNameNode分别是干什么的？

3. Yarn工作原理

4. hdfs的工作流程

读流程：

写流程

5.block块副本放置策略

combiner和partition的作用：

namenode的safemode是怎么回事？如何才能退出safemode？

一个datanode 宕机,怎么一个流程恢复？

Hadoop宕机

hadoop 的 namenode 宕机,怎么解决？

Hadoop数据倾斜问题？

Hadoop常用端口号

hadoop配置文件

集群搭建流程

yarn的默认调度器，调度器分类，以及他们之间的区别

Hadoop的优化

面对许多小文件

Hadoop参数调优

Spark常见的面试题

RDD是什么？创建方式？分别怎么分区的？

2. task stage job分别是什么？

3. RDD的缓存持久机制

4. spark常见调优

HIve

hive内部表和外部表的区别

分桶的作用

Hive 你们用的是外部表还是内部表,有没有写过UDF。

KAFKA

1.Kafka中的ISR(InSyncRepli)、OSR(OutSyncRepli)、AR(AllRepli)代表什么？

Kafka种的HW,LEO等分别代表什么？

Kafka的用途有哪些？使用场景如何？

kafka怎样体现消息的顺序性的？

消费组中的消费者个数如果超过topic的分区，那么就会有消费者消费不到数据这句话是否正确？

有哪些情形会造成重复消费，或丢失信息？

KAFKA分区的目的

KAFKA的高可靠性是怎么实现的？

topic的分区数目可不可以增加？

topic分区数可不可以减少？

简述Kafka的日志目录结构

如何解决消费者速率低的问题

kafka有哪些设计让他有如此高的性能

kafka启动不起来的原因

kafkaController的作用

失效副本是什么？ 有哪些应对的措施？

Kafka消息采用的是Pull模式，还是Push模式？

Kafka创建Topic时如何将分区放置到不同的broker中?

kafka中的事务是怎么实现的

Kafka中的分区器、序列化器、拦截器是否了解？它们之间的处理顺序是什么？

Kafka生产者客户端的整体结构是什么样子的？使用了几个线程来处理？分别是什么？

消费者提交消费位移时提交的是当前消费到的最新消息的offset还是offset+1？

Kafka 的机器数量

Kafka压测

Kafka 的日志保存时间

Kafka的硬盘大小

Kafka监控

Kafka分区分配策略

Kafka 消息数据积压，Kafka 消费能力不足怎么处理？

Kafka的幂等性

Kafka事务

Kafka高效读写数据

Kafka 挂掉

Linux&Shell

linux常用命令

shell常用命令

Flume

1Flume组成，Put事务，Take事务

2Flume拦截器

3FlumeChannel拦截器

4Flume监控器

5Flume采集会数据丢失么?

6Flume内存

7FileChannel优化

位图和布隆

线性回归算法

贝叶斯

失效副本是什么？有哪些应对的措施？