双非菜鸡的秋招结束~

在牛客看了好多，发个帖子总结一下

全是个人总结，如有错误敬请谅解😁😁

如有错误，虚心接受各位大佬赐教更改😁😁

——————————————————————————————————————

1、Flume

• flume是一个分布式、可靠和高可用的海量日志采集、聚合和传输的系统

• flume的管道是基于事务的，保证了数据在传送和接收时的一致性

• 一个flume进程包括：Source（数据源） Channel（处理介质） Sink（目的地）

• flume的管道（Channel自带两种）可以使用内存（Memory Channel），也可以使用文件方式(File Channel)来防止宕机

• flume的最主要作用是可以实时读取服务器的日志数据

2、Spark

• spark是什么：Spark是基于内存的迭代式计算引擎

• Spark的运行原理以及什么时候发生宽窄依赖：

  • 运行原理：（客户端）Spark通过SparkContex向集群的Master提交执行作业的请求，Maste节点的资源管理器申请执行所需的CPU，内存等，然后Master节点在worker启动Executorj进程，然后向客户端反向注册，客户端会根据DAG有向无环图来划分Stage，为每一个stage创建一个task，并且封装至TaskSchedual中，然后讲task交给Executor去执行，在action算子执行之后将结果收集到客户端。

  • 宽窄依赖：RDD转换的时候发生宽窄依赖

  • Spark中join算子可能会发生宽依赖也可能不会

• Spark的执行模式：

  • Local 本地执行，通过多线程实现并行计算

  • Standalone 是 Spark 提供的一种内置的集群模式，采用内置的资源管理器进行管理

  • Spark on Yarn模式：Spark 支持将作业提交到 Yarn 上运行，此时不需要启动 Master 节点，也不需要启动 Worker 节点

    • 需要配置在 spark-env.sh

    • YARN_CONF_DIR=/usr/app/hadoop-2.6.0-cdh5.15.2/etc/hadoop # JDK安装位置 JAVA_HOME=/usr/java/jdk1.8.0_201

      启动时必须保证Hadoop已经启动了

• RDD的属性： RDD叫做分布式数据集，是spark中最基本的数据抽象，它代表一个不可变、可分区、里边的元素可并行计算的集合，RDD的属性： 一组分区|分片、应用在分区上的函数、分区间的依赖关系、分区函数（对key-value的RDD才有）、存放每个分区的位置列表

• RDD的两种操作（算子）： 转换算子（Transformation） 动作算子（Action）

• DStream和DStreamGraph的区别：在创建StreamingContext的时候会创建DStreamGraph

  • Spark Streaming里的DStream可以看成是Spark Core里的RDD的模板，DStreamGraph是RDD DAG的模板（是基于其方法中的inputStream和outputStream实现的）

• Spark重新分区：

  • repartition(numPartitions:Int):RDD[T]

  • coalesce(numPartitions:Int，shuffle:Boolean=false):RDD[T]

  • 它们两个都是RDD的分区进行重新划分，repartition只是coalesce接口中shuffle为true的简易实现，（假设RDD有N个分区，需要重新划分成M个分区

    • N<M，一般情况下N个分区有数据分布不均匀的状况，利用HashPartitioner函数将数据重新分区为M个，这时需要将shuffle设置为true

    • 如果N>M并且N和M相差不多，(假如N是1000，M是100)那么就可以将N个分区中的若干个分区合并成一个新的分区，最终合并为M个分区，这时可以将shuff设置为false，在shuffl为false的情况下，如果M>N时，coalesce为无效的

3、Kafka

• kafka的（控制器Broker）选举机制：Kakfa Broker Leader的选举：Kakfa Broker集群受Zookeeper管理。所有的Kafka Broker节点一起去Zookeeper上注册一个临时节点，因为只有一个Kafka Broker会注册成功，其他的都会失败，所以这个成功在Zookeeper上注册临时节点的这个Kafka Broker会成为Kafka Broker Controller，其他的Kafka broker叫Kafka Broker follower。（这个过程叫Controller在ZooKeeper注册Watch）。

• kafka的分区副本选举机制：Kafka在Zookeeper上针对每个Topic都维护了一个ISR（in-sync replica---已同步的副本）的集合，集合的增减Kafka都会更新该记录。如果某分区的Leader不可用，Kafka就从ISR集合中选择一个副本作为新的Leader。

• 在kafka的集群中，会存在着多个主题topic，在每一个topic中，又被划分为多个partition，为了防止数据不丢失，每一个partition又有多个副本，在整个集群中，总共有三种副本角色：

  首领副本（leader）：也就是leader主副本，每个分区都有一个首领副本，为了保证数据一致性，所有的生产者与消费者的请求都会经过该副本来处理。 跟随者副本（follower）：除了首领副本外的其他所有副本都是跟随者副本，跟随者副本不处理来自客户端的任何请求，只负责从首领副本同步数据，保证与首领保持一致。如果首领副本发生崩溃，就会从这其中选举出一个leader。 首选首领副本：创建分区时指定的首选首领。如果不指定，则为分区的第一个副本。

• kafka消费者组选主：在kafka的消费端，会有一个消费者协调器以及消费组，组协调器GroupCoordinator需要为消费组内的消费者选举出一个消费组的leader，如果消费组内还没有leader，那么第一个加入消费组的消费者即为消费组的leader，如果某一个时刻leader消费者由于某些原因退出了消费组，那么就会重新选举leader。

• kafka传输的三种事务定义：（1）最多一次: 消息不会被重复发送，最多被传输一次，但也有可能一次不传输

  （2）最少一次: 消息不会被漏发送，最少被传输一次，但也有可能被重复传输.

  （3）精确的一次（Exactly once）: 不会漏传输也不会重复传输,每个消息都传输被一次而且仅仅被传输一次，这是大家所期望的

• kafka的幂等性：为了实现Producer的幂等性，Kafka引入了Producer ID（即PID）和Sequence Number。

  • PID：每个新的Producer在初始化的时候会被分配一个唯一的PID，这个PID对用户是不可见的。

• Sequence Numbler：（对于每个PID，该Producer发送数据的每个<Topic, Partition>都对应一个从0开始单调递增的Sequence Number

  • Kafka可能存在多个生产者，会同时产生消息，但对Kafka来说，只需要保证每个生产者内部的消息幂等就可以了，所有引入了PID来标识不同的生产者。

• 对于Kafka来说，要解决的是生产者发送消息的幂等问题。也即需要区分每条消息是否重复。 Kafka通过为每条消息增加一个Sequence Numbler，通过Sequence Numbler来区分每条消息。每条消息对应一个分区，不同的分区产生的消息不可能重复。所有Sequence Numbler对应每个分区

• Broker端在缓存中保存了这seq number，对于接收的每条消息，如果其序号比Broker缓存中序号大于1则接受它，否则将其丢弃

4、kafka如何保证生产者不丢失数据，消费者不丢失数据

• 生产者：

  • kafka的ack机制：在kafka发送数据的时候，每次发送消息都会有一个确认反馈机制，确保消息正常的能够被收到，其中状态有0,1，-1

  • 如果是0 ，那么代表发送过去，不等待kafka消息确认，认为成功，一定会丢失消息，可能kafka集群正在选举，此时就无法收到任何异常

  • 如果是1，那么代表发送过去，等待首领副本确认消息，认为成功首领肯定收到了消息，写入了分区文件（不一定落盘）

  • 如果是all, 那么代表发送过去之后，消息被写入所有同步副本之后 ，认为成功

  • 注意这里是 所有同步副本，不是所有副本。 具体是多少同步副本，还要取决于kafka集群设置的最小同步副本数，和集群当前的同步副本数

  • 选择这种配置，会可靠，但是牺牲效率，可以通过，增大批和使用异步模式，提高效率

  • 异步模式下的有个buffer，通过buffer来进行控制数据的发送，有两个值来进行控制，时间阈值与消息的数量阈值，如果buffer满了数据还没有发送出去，有个选项是配置是否立即清空buffer。可以设置为-1，永久阻塞，也就数据不再生产

• 消费者：

  • 通过offset commit 来保证数据的不丢失，kafka自己记录了每次消费的offset数值，下次继续消费的时候，会接着上次的offset进行消费

• Broker端：

  • 设置unclean.leader.election.enable为false，禁止非ISR列表中的副本参与分区首领副本选举

5、Zookeeper的消息的发布订阅功能

• 数据的发布/订阅系统，通常也用作配置中心。在分布式系统中，你可能有成千上万个服务节点，如果想要对所有服务的某项配置进行更改，由于数据节点过多，你不可逐台进行修改，而应该在设计时采用统一的配置中心。之后发布者只需要将新的配置发送到配置中心，所有服务节点即可自动下载并进行更新，从而实现配置的集中管理和动态更新

• Zookeeper 通过 Watcher 机制可以实现数据的发布和订阅。分布式系统的所有的服务节点可以对某个 ZNode 注册监听，之后只需要将新的配置写入该 ZNode，所有服务节点都会收到该事件

6、Zookeeper的功能

• maste选举：

  • 分布式系统一个重要的模式就是主从模式 (Master/Salves)，Zookeeper 可以用于该模式下的 Matser 选举。可以让所有服务节点去竞争性地创建同一个 ZNode，由于 Zookeeper 不能有路径相同的 ZNode，必然只有一个服务节点能够创建成功，这样该服务节点就可以成为 Master 节点

• 分布式锁：

  • 可以通过 Zookeeper 的临时节点和 Watcher 机制来实现分布式锁，这里以排它锁为例进行说明：

    • 分布式系统的所有服务节点可以竞争性地去创建同一个临时 ZNode，由于 Zookeeper 不能有路径相同的 ZNode，必然只有一个服务节点能够创建成功，此时可以认为该节点获得了锁。其他没有获得锁的服务节点通过在该 ZNode 上注册监听，从而当锁释放时再去竞争获得锁。锁的释放情况有以下两种：

      • 当正常执行完业务逻辑后，客户端主动将临时 ZNode 删除，此时锁被释放；

      • 当获得锁的客户端发生宕机时，临时 ZNode 会被自动删除，此时认为锁已经释放。

      当锁被释放后，其他服务节点则再次去竞争性地进行创建，但每次都只有一个服务节点能够获取到锁，这就是排他锁。

• 集群管理：

  • 如可以通过创建临时节点来建立心跳检测机制。如果分布式系统的某个服务节点宕机了，则其持有的会话会超时，此时该临时节点会被删除，相应的监听事件就会被触发

  • 分布式系统的每个服务节点还可以将自己的节点状态写入临时节点，从而完成状态报告或节点工作进度汇报

  • 通过数据的订阅和发布功能，Zookeeper 还能对分布式系统进行模块的解耦和任务的调度

  • 通过监听机制，还能对分布式系统的服务节点进行动态上下线，从而实现服务的动态扩容

• zookeeper特性：

  • 顺序一致性：从一个客户端发起的事务请求，最终都会严格按照其发起顺序被应用到 Zookeeper 中

  • 原子性：所有事务请求的处理结果在整个集群中所有机器上都是一致的；不存在部分机器应用了该事务，而另一部分没有应用的情况

  • 单一视图：所有客户端看到的服务端数据模型都是一致的

  • 可靠性：一旦服务端成功应用了一个事务，则其引起的改变会一直保留，直到被另外一个事务所更改

  • 实时性：一旦一个事务被成功应用后，Zookeeper 可以保证客户端立即可以读取到这个事务变更后的最新状态的数据

7、Spark中的shuffle机制

• 概念：Shuffle就是对数据进行重组，由于分布式计算的特性和要求，在实现细节上更加繁琐和复杂

• Shuffle Write：上一个stage的每一个map task就必须保证将自己处理的当前分区中的数据相同的key写入一个分区文件中，可能会写入多个不同的分区文件中

• Shuffle Read：reduce task就会从上一个stage的所有task所在的机器上寻找属于自己的那些分区文件，这样就可以保证每一个key所对应的value都会汇聚在同一个节点上去处理和聚合

8、Spark与hadoop的区别

• hadoop是一个生态圈，包含了hdfs（hadoop分布式文件系统），yarn（资源调度平台），mapreduce（基于yarn的离线计算框架），而Spark是基于内存的大数据并行计算框架，对标于MR

• Spark并没有提供文件管理系统，它必须和其他的分布式文件系统进行集成才可以运作

• Spark处理数据的设计模式与MR不同，MR在处理数据时涉及到多次磁盘IO，而Spark的设计模式是读取集群中的数据后在内存中进行存储计算，直到运算完毕后再存储至集群中

• 与Hadoop的MapReduce相比，Spark基于内存的运算要快100倍以上，基于硬盘的运算也要快10倍以上

• Spark通过RDD实现了对迭代计算

• Spark有scala语言编写，而Hadoop由java语言编写

9、Hadoop里的namenode和datanode

• namenode：

  • 是整个文件系统的管理节点，它维护着整个文件系统的文件目录树，文件/目录的元信息和每个文件对应的数据块列表

  • 接收用户的操作请求，同时在namenode上保存着：fsimage: 元数据镜像文件。存储某一时段NameNode内存储的元数据信息（序列化生成镜像文件）edits:操作日志文件 fstime:保存最近一次checkpoint的时间

• datanode：提供真实的文件数据存储服务（再datanode中存储block数据块）

  • 数据块：hdfs中的基本存储单元 不同于普通文件系统的是，HDFS中，如果一个文件小于一个数据块的大小，并不占用整个数据块存储空间 数据块通过设置副本机制，默认是3个副本

10、spark的lazy体现在哪里

• Spark中，Transformation方法都是懒操作方法，比如map,flatMap,reduceByKey等。当触发某个Action操作时才真正执行

• Spark通过lazy特性，可以进行底层的spark应用执行的优化。在生活中，就像三思而后行。谋定而后动

• 懒加载的意义：

  • 不运行job就触发计算，避免了大量的无意义的计算，即避免了大量的无意义的中间结果的产生，即避免产生无意义的磁盘I/O及网络传输

  • 更深层次的意义在于，执行运算时，看到之前的计算操作越多，执行优化的可能性就越高

11、Spark中数据倾斜

• 数据倾斜：在任务执行期间，RDD会被分为一系列的分区，每个分区都是整个数据集的子集。当spark调度并运行任务的时候，Spark会为每一个分区中的数据创建一个任务。大部分的任务处理的数据量差不多，但是有少部分的任务处理的数据量很大，因而Spark作业会看起来运行的十分的慢，从而产生数据倾斜（进行shuffle的时候）

• 数据倾斜只出现在shuffle过程中，可能会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等

• 解决办法：

  • 过滤少量导致数据倾斜的key

  • 对key添加随机前缀，在执行操作后再去掉前缀

  • 自定义Partitioner

12、Yarn是什么

• yarn是Hadoop中的资源调度平台

• yarn中的角色：

  • ResourceManager：负责接收任务，管理集群中的资源和调度，以及监控运行再yarn上的程序

  • NodeManager：真正执行计算的节点

  • ApplicationMaster：一个job对应一个application master，创建task，启动yarnchild进程，并且分配task给yarnchild去执行

  • YarnChild：真正执行task的进程

13、hdfs读写流程

• hdfs写：

  • 首先客户端给NameNode发送上传文件的请求（包含文件名，文件大小）

  • NameNode的给客户端响应，如果文件已经存在则报错，不存在则继续执行

  • 客户端将要上传的文件划分为block块并且请求上传第一个block（128mb）

  • NameNode以Pipeline的模式返回可上传文件的DataNode的列表

  • 客户端连接DataNode进行block块的上传

• hdfs读：

  • 客户端将要读取的文件路径发送给namenode，namenode获取文件的元信息（主要是block的存放位置信息）返回给客户端，客户端根据返回的信息找到相应datanode逐个获取文件的block并在客户端本地进行数据追加合并从而获得整个文件

14、Hadoop的特性

• 高可靠性：采用冗余数据存储方式，即使一个副本发生故障，其他副本也可以保证对外提供服务

• 高效性：采用分布式存储和分布式处理两大核心你技术，能够高效的处理pb级数据

• 高可扩展性：Hadoop的设计目标是可以高效稳定的运行在廉价的计算机群上，可以扩展到数以千计的计算机节点上

• 高容错性：采用冗余数据存储方式，自动保存数据的多个副本，并且能够自动将失败的任务进行重新分配

• 成本低：Hadoop采用廉价的计算机群，成本较低

• 运行在Linux平台上：Hadoop是基于java语言开发的，可以很好的运行在Linux平台上

• 支持多种编程语言：Hadoop的应用程序可以使用多种语言编写

15、Hadoop如何恢复被删除的文件

• hadoop的hdfs中被删除文件的恢复原理和回收站原理是一样的，就是在删除hdfs文件时，被删除的文件被移动到了hdfs的.Trash文件夹中，恢复时只需将该文件夹中文件拿出即可。具体操作如下：

• 设置.Trash文件夹

  如果需要恢复hdfs中文件，就需要设置.Trash，hadoop的.Trash默认是关闭的。具体设置如下   <property>  <name>fs.trash.interval</name>  <value>10080</value>  </property>

  该配置项在core-site.xml中，fs.trash.interval代表删除的文件保留的时间，时间单位为分钟，默认为0代表不保存删除的文件。我们只需要设置该时间即可打开.Trash

• 设置后删除文件会显示删除的文件被移动到了hdfs://192.168.1.100:9000/user/hadoop/.Trash/Current中

• 恢复时只需要将.Trash中文件移动到我们设置的目录即可

16、Spark的stage划分

• Spark任务会根据RDD之间的依赖关系，形成一个DAG有向无环图，DAG会提交给DAGScheduler，DAGScheduler会把DAG划分相互依赖的多个stage，划分stage的依据就是RDD之间的宽窄依赖。遇到宽依赖就划分stage,每个stage包含一个或多个task任务。然后将这些task以taskSet的形式提交给TaskScheduler运行。 stage是由一组并行的task组成。

• stage切割规则：从后往前，遇到宽依赖就切割stage

17、Spark为什么快

• 消除了冗余的HDFS读写：Hadoop每次shuffle操作后，必须写到磁盘，而Spark在shuffle后不一定落盘，可以cache到内存中，以便迭代时使用。如果操作复杂，很多的shufle操作，那么Hadoop的读写IO时间会大大增加

• 消除了冗余的MapReduce阶段：Hadoop的shuffle操作一定连着完整的MapReduce操作，冗余繁琐。而Spark基于RDD提供了丰富的算子操作，且reduce操作产生shuffle数据，可以缓存在内存中

• JVM 的优化: Hadoop 每次 MapReduce 操作，启动一个 Task 便会启动一次 JVM，基于进程的操作。而 Spark 每次 MapReduce 操作是基于线程的，只在启动 Executor 时启动一次 JVM，内存的 Task 操作是在线程复用的。每次启动 JVM 的时间可能就需要几秒甚至十几秒，那么当 Task 多了，这个时间 Hadoop 不知道比 Spark 慢了多少

18、Spark有关资源分配的参数

• 分配的资源：

  • 分配的 executor 数量

  • 每个 executor 需要的 core 数量

  • 每个 executor 需要的内存大小

  • driver 的内存大小 (这个影响不大)

  • 在spark提交作业的时候可以对这些资源进行配置

• Spark在集群模式下的运行：

  • 构建Spark Application（就是提交的Spark运行程序）的运行环境（启动SparkContext），SparkContext向资源管理器（可以是Standalone、Mesos或YARN）注册并申请运行Executor资源

  • 资源管理器分配Executor资源，并启动监听程序，Executor运行情况将随着心跳发送到资源管理器上

  • SparkContext构建成DAG图，将DAG图分解成Stage，并把Task发送给Task Scheduler，Executor向SparkContext申请Task

  • Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor

  • Task在Executor上运行，运行完毕释放所有资源

• 资源分配的两种方式：

  • 静态分配:在提交任务的时候设置

  • 动态分配:根据数据的大小、需要的运算能力设置，便于以后类似任务的重用

19、Kafka为什么具有高吞吐量低延迟的特点

• 传统的读取文件数据并发送到网络：

  • 操作系统将数据从磁盘文件中读取到内核空间的页面缓存

  • 应用程序将数据从内核空间读入用户空间缓冲区

  • 应用程序将读到数据写回内核空间并放入socket缓冲区

  • 操作系统将数据从socket缓冲区复制到网卡接口，此时数据才能通过网络发送

  • “零拷贝技术”只用将磁盘文件的数据复制到页面缓存中一次，然后将数据从页面缓存直接发送到网络中（发送给不同的订阅者时，都可以使用同一个页面缓存），避免了重复复制操作

• 实际上不管是内存还是磁盘，快或慢关键在于寻址的方式，磁盘分为顺序读写与随机读写，内存也一样分为顺序读写与随机读写

• kafka是将消息记录持久化到本地磁盘中的，而kafka基于磁盘的持久化是基于硬盘的顺序读写的，而基于磁盘的顺序存储的速度甚至超过基于内存的随机读写，kafka的message是不断追加到本地磁盘文件末尾的

• 为了优化读写性能，Kafka利用了操作系统本身的Page Cache，就是利用操作系统自身的内存而不是JVM空间内存，通过操作系统的Page Cache，Kafka的读写操作基本上是基于内存的，读写速度得到了极大的提升

• 零拷贝：

  • linux操作系统 “零拷贝” 机制使用了sendfile方法， 允许操作系统将数据从Page Cache 直接发送到网络，只需要最后一步的copy操作将数据复制到 NIC 缓冲区， 这样避免重新复制数据

  • 通过这种 “零拷贝” 的机制，Page Cache 结合 sendfile 方法，Kafka消费端的性能也大幅提升

  • 零拷贝主要是避免了在内核空间和用户空间之间的拷贝

• 批量读写：kafka在应用程序层面提供的技术，在写入数据时可以启用批次写入

• 分区分段+索引：

  • Kafka的message是按topic分类存储的，topic中的数据又是按照一个一个的partition即分区存储到不同broker节点。每个partition对应了操作系统上的一个文件夹，partition实际上又是按照segment分段存储的。这也非常符合分布式系统分区分桶的设计思想

  • 通过这种分区分段的设计，Kafka的message消息实际上是分布式存储在一个一个小的segment中的，每次文件操作也是直接操作的segment。为了进一步的查询优化，Kafka又默认为分段后的数据文件建立了索引文件，就是文件系统上的.index文件。这种分区分段+索引的设计，不仅提升了数据读取的效率，同时也提高了数据操作的并行度

• 批量压缩：

  • Kafka速度的秘诀在于，它把所有的消息都变成一个批量的文件，并且进行合理的批量压缩，减少网络IO损耗，提高I/O速度，写入数据的时候由于单个Partion是末尾添加所以速度最优；读取数据的时候配合sendfile直接暴力输出

20、Hbase中Region分裂

• 当MemStore的数据超过阈值时，将数据溢写磁盘，生成一个StoreFile文件。当Region中最大Store的大小超过阈值时，Region分裂，等分成两个Region，实现数据访问的负载均衡。新的Region的位置由HMaster来确定在哪个RegionServer中

• Region分裂过程：

  • RegionServer决定本地的region分裂，第一步是，在zookeeper的/hbase/region-in-reansition/region-name下创建一个znode

  • Master通过父region-in-transition znode的watcher监测到刚刚创建的znode

  • RegionServer在HDFS中父region的目录下创建名为“.split”的子目录

  • 关闭要分裂的region的读写请求

  • 在split目录中创建所需要的文件结构

  • 移动两个新region文件目录到目录表中，并更新.META表

  • 异步复制父region中的数据到两个子region中(最主要的消耗时间)

  • 复制完成后、打开两个新产生的region，并上线，可以接收新的读写请求

  • 异步任务把定时把原来被分裂的region从.META表中清除掉，并从文件系统中删除该region所占用的空间

• 分割点的确定： Region中StoreFile最大文件中的内部中点

21、Zookeeper相关

• zookeeper的节点类型：

  • PERSISTENT：持久化节点

  • PERSISTENT_SEQUENTIAL：持久化有序节点

  • PHEMERAL：临时节点

  • EPHEMERAL_SEQUENTIAL：临时有序节点

• zookeeper数据存储：

  • ZooKeeper的数据模型是一棵树，而从使用角度看， Zookeeper就像一个内存数据库一样。在这个内存数据库中，存储了整棵树的内容，包括所有的节点路径、节点数据及其ACL信息等，Zookeeper会定时将这个数据存储到磁盘上

  • DataTree：DataTree是内存数据存储的核心，是一个树结构，代表了内存中一份完整的数据。DataTree不包含任何与网络、客户端连接及请求处理相关的业务逻辑，是一个独立的组件

  • DataNode是数据存储的最小单元，其内部除了保存了结点的数据内容、ACL列表、节点状态之外，还记录了父节点的引用和子节点列表两个属性，其也提供了对子节点列表进行操作的接口

  • ZKDatabase：Zookeeper的内存数据库，管理Zookeeper的所有会话、DataTree存储和事务日志。ZKDatabase会定时向磁盘dump快照数据，同时在Zookeeper启动时，会通过磁盘的事务日志和快照文件恢复成一个完整的内存数据库

• 目录结构：Bin 主要的一些执行命令 Conf 存放配置文件，需要修改zk.cfg Contrib 附加的一些功能

  Dist-maven Mvn编译后的目录 Docs文档 Lib 需要依赖的jar包 Src 源文件

• leader选举如何实现：投票选举，id大的优先获得投票

22、Spark中的Stage、Job、Task的划分

• Job的划分：根据Action算子进行划分，一个Action算子划分一个Job

• Stage的划分：按照DAG图，从后往前，遇到宽依赖就进行Stage的划分

• Task的划分：一个Stage中，最后一个RDD有多少个partition就划分多少个task

23、Kafka分区的目的及有序性

• 分区对于 Kafka 集群的好处是：实现负载均衡

• 分区对于消费者来说：可以提高并发度，提高效率

• Kafka如何做到消息有序性的：

  • kafka是无法保证全局的消息顺序性的，只能保证主题的某个分区的消息顺序性

  • kafka 中的每个 partition 中的消息在写入时都是有序的，而且单独一个 partition 只能由一个消费者去消费，可以在里面保证消息的顺序性。但是分区之间的消息是不保证有序的

• kafka如何保证数据的顺序消费：

  • kafka的顺序消息仅仅是通过partitionKey，将某类消息写入同一个partition，一个partition只能对应一个消费线程，以保证数据有序，除了发送消息需要指定partitionKey外，producer和consumer实例化无区别

  • 生产者在写的时候，可以指定一个 key，比如说我们指定了某个订单 id 作为 key，那么这个订单相关的数据，一定会被分发到同一个 partition 中去，而且这个 partition 中的数据一定是有顺序的

• 多线程进行消息消费如何保证顺序：

  • 写 N 个内存 queue，具有相同 key 的数据都到同一个内存 queue；然后对于 N 个线程，每个线程分别消费一个内存 queue 即可，这样就能保证顺序性

24、Kafka消息是采用Push模式还是Pull模式

• Kafka最初考虑的问题是，customer应该从brokes拉取消息还是brokers将消息推送到consumer，也就是pull还push。在这方面，Kafka遵循了一种大部分消息系统共同的传统的设计：producer将消息推送到broker，consumer从broker拉取消息

• 一些消息系统比如Scribe和Apache Flume采用了push模式，将消息推送到下游的consumer。这样做有好处也有坏处：由broker决定消息推送的速率，对于不同消费速率的consumer就不太好处理了。消息系统都致力于让consumer以最大的速率最快速的消费消息，但不幸的是，push模式下，当broker推送的速率远大于consumer消费的速率时，consumer恐怕就要崩溃了。最终Kafka还是选取了传统的pull模式

• Pull模式的另外一个好处是consumer可以自主决定是否批量的从broker拉取数据。Push模式必须在不知道下游consumer消费能力和消费策略的情况下决定是立即推送每条消息还是缓存之后批量推送。如果为了避免consumer崩溃而采用较低的推送速率，将可能导致一次只推送较少的消息而造成浪费。Pull模式下，consumer就可以根据自己的消费能力去决定这些策略。 Pull有个缺点是，如果broker没有可供消费的消息，将导致consumer不断在循环中轮询，直到新消息到t达。为了避免这点，Kafka有个参数可以让consumer阻塞知道新消息到达(当然也可以阻塞知道消息的数量达到某个特定的量这样就可以批量发送）

25、Kafka中的ZooKeeper是什么？是否可以脱离ZooKeeper独立运行？

• Zookeeper是一个开放源码的、高性能的协调服务，它用于Kafka的分布式应用

• 不可以，不可能越过Zookeeper直接联系Kafka broker，一旦Zookeeper停止工作，它就不能服务客户端请求

• Zookeeper主要用于在集群中不同节点之间进行通信，在Kafka中，它被用于提交偏移量，因此如果节点在任何情况下都失败了，它都可以从之前提交的偏移量中获取，除此之外，它还执行其他活动，如: leader检测、分布式同步、配置管理、识别新节点何时离开或连接、集群、节点实时状态等等

26、ZooKeeper的ZAB协议

• ZAB 协议是 Zookeeper 专门设计的一种支持崩溃恢复的原子广播协议。通过该协议，Zookeepe 基于主从模式的系统架构来保持集群中各个副本之间数据的一致性

• Zookeeper 使用一个单一的主进程来接收并处理客户端的所有事务请求，并采用原子广播协议将数据状态的变更以事务 Proposal 的形式广播到所有的副本进程上

• ZAB协议包括两种基本的模式：

  • 崩溃恢复

    • 当整个服务框架在启动过程中，或者当 Leader 服务器出现异常时，ZAB 协议就会进入恢复模式，通过过半选举机制产生新的 Leader，之后其他机器将从新的 Leader 上同步状态，当有过半机器完成状态同步后，就退出恢复模式，进入消息广播模式

  • 消息广播

    • ZAB 协议的消息广播过程使用的是原子广播协议。在整个消息的广播过程中，Leader 服务器会每个事物请求生成对应的 Proposal，并为其分配一个全局唯一的递增的事务 ID(ZXID)，之后再对其进行广播

27、Spark中的常用算子

• 窄依赖：子RDD中一个分区中的数据来自于父RDD一个分区中的数据

• 宽依赖：子RDD中一个分区中的数据来自于父RDD中的多个分区中的数据

• union：合并两个RDD，窄依赖

• join：在一个 (K, V) 和 (K, W) 类型的 Dataset 上调用时，返回一个 (K, (V, W)) 的 Dataset，等价于内连接操作。如果想要执行外连接，可以使用 leftOuterJoin, rightOuterJoin 和 fullOuterJoin 等算子

  • 可能会发生宽依赖

  • 也可能会发生窄依赖

• 会发生shuffle的算子： distinct() reduceByKey() groupBy() groupByKey() sortByKey() sortBy()

• foreach：是对RDD中的每一个元素执行指定的函数

• foreachPartition：是对RDD中的每一个Partition执行指定的函数

大数据部分2

1、Kafka如何保证数据不丢失

• Producer端：

  • 如果设置ack=all，则生产者端不会产生数据丢失，会不断重试直至成功

  • 选择这种配置，会可靠，但是牺牲效率，可以通过，增大批和使用异步模式，提高效率

  • 异步模式下的有个buffer，通过buffer来进行控制数据的发送，有两个值来进行控制，时间阈值与消息的数量阈值，如果buffer满了数据还没有发送出去，有个选项是配置是否立即清空buffer。可以设置为-1，永久阻塞，也就数据不再生产

• Broker端：

  • 在producer端设置ack=all，等待数据写入所有同步副本才算成功

  • 为每个分区维护至少两个同步副本（ISR列表中至少有两个）

• Consumer：

  • 关闭自动更新offset，等到数据被处理后再手动跟新offset

  • 在消费前做验证前拿取的数据是否是接着上回消费的数据，不正确则return先行处理排错

2、ZooKeeper是如何保证数据一致性的

• ZAB 协议是 Zookeeper 专门设计的一种支持崩溃恢复的原子广播协议。通过该协议，Zookeepe 基于主从模式的系统架构来保持集群中各个副本之间数据的一致性

• zookeeper使用一个单一的主进程来接收并处理客户端的所有事务请求，并采用原子广播协议将数据状态的变更以事务Proposal的形式广播到所有的副本进程上去，以此来保证数据的一致性

• 具体流程：所有的事务请求必须由唯一的Leader服务来处理，Leader服务将事务请求转换为事务Proposal，并将该Proposal分发给集群中所有的Follower服务，如果有半数的Follower服务进行了正确反馈，那么Leader服务就会再次向所有的Follower发出Commit消息，要求将前一个Proposal进行提交

3、Spark的内存模型

• Spark在一个Executor中的内存分为三块，一块是execution内存，一块是storage内存，一块是other内存

  • execution内存是执行内存，map、join等都是在这块内存中执行，shuffle的数据也会先缓存在这个内存中，满了再写入磁盘中，可以减少IO

  • storage内存是存储broadcast、cache、persist数据的地方

  • other内存是Exeuctor程序执行时预留给自己的内存

• 在Spark-1.6.0后加入了堆外内存，进一步优化了Spark的内存使用，堆外内存使用JVM堆以外的内存，不会被gc回收，可以减少频繁的full gc，所以在Spark程序中，会长时间逗留再Spark程序中的大内存对象可以使用堆外内存存储

4、Spark中的OOM（内存溢出）问题

• OOM问题通常就出现在execution内存中

• OOM问题的两种情况：

  • map执行中内存溢出

    • map执行中内存溢出代表了所有map类型的操作，包括：flatMap，filter，mapPatitions等

  • shuffle后内存溢出

    • shuffle后内存溢出的shuffle操作包括join，reduceByKey，repartition等操作

• 内存溢出的解决办法：

  • map过程产生大量对象导致内存溢出：

    • 在不增加内存的情况下，可以通过减少每个Task的大小，以便达到每个Task即使产生大量的对象Executor的内存也能够装得下，具体做法可以在会产生大量对象的map操作之前调用repartition方法，分区成更小的块传入map

  • 数据不平衡导致的内存溢出

    • 与map过程的相同，也是调用repartition进行重新分区

  • shuffle后的内存溢出：

    • shuffle内存溢出的情况可以说都是shuffle后，单个文件过大导致的，如果是别的partitioner导致的shuffle内存溢出，就需要从partitioner的代码增加partitions的数量

  • standalone模式下资源分配不均匀导致内存溢出：

    • 在standalone的模式下如果配置了--total-executor-cores 和 --executor-memory 这两个参数，但是没有配置--executor-cores这个参数的话，就有可能导致，每个Executor的memory是一样的，但是cores的数量不同，那么在cores数量多的Executor中，由于能够同时执行多个Task，就容易导致内存溢出的情况。这种情况的解决方法就是同时配置--executor-cores或者spark.executor.cores参数，确保Executor资源分配均匀

  • 在RDD***用对象能够减少OOM的情况

5、Spark中的Shuffle机制

• 一个 RDD 由一个或者多个分区（Partitions）组成。对于 RDD 来说，每个分区会被一个计算任务所处理，用户可以在创建 RDD 时指定其分区个数，如果没有指定，则默认采用程序所分配到的 CPU 的核心数

• 在 Spark 中，一个任务对应一个分区，通常不会跨分区操作数据。但如果遇到 reduceByKey 等操作，Spark 必须从所有分区读取数据，并查找所有键的所有值，然后汇总在一起以计算每个键的最终结果 ，这称为 Shuffle

• Shuffle Write：上一个stage的每一个map task就必须保证将自己处理的当前分区中的数据相同的key写入一个分区文件中，可能会写入多个不同的分区文件中

• Shuffle Read：reduce task就会从上一个stage的所有task所在的机器上寻找属于自己的那些分区文件，这样就可以保证每一个key所对应的value都会汇聚在同一个节点上去处理和聚合

• 会导致shuffle发生的算子：

  • 涉及到重新分区操作： 如 repartition 和 coalesce

  • 所有涉及到 ByKey 的操作：如 groupByKey 和 reduceByKey，但 countByKey 除外

  • 联结操作：如 cogroup 和 join

6、YARN中的资源调度器

• 在yarn中负责给应用分配资源的Scheduler，共有三种资源调度器

  • FIFO Scheduler：把应用按照提交的顺序排成一个 先进先出 的队列，缺点是大的应用可能会占用集群的所有资源，导致其他应用被阻塞

  • Capacity Scheduler：容量调度器，以队列的形式配置集群资源，每个队列可以抢占队列资源（可以设置最大容量），多个队列可以同时执行任务，但每一个队列内部还是FIFO

  • Fair Schduler：公平调度器，同样的以队列的形式配置集群资源，每个队列可以抢占资源，当被抢占的队列有任务时，抢占的资源需要奉还（用完在奉还）（每个队列中的job都会分配到资源以确保公平）

7、RDD 中的 reduceByKey 与 groupByKey 哪个性能高

• reduceByKey在进行网络传输之前首先会在每个分区上对相同key的数据进行聚合，然后在传输后再次进行聚合

• groupByKey会将分区中所有的键值对经过网络传输之后再进行聚合

• 所以在传输大数据集的时候reduceByKey的性能大于groupByKey

8、Flume中的事务

• 在Flume中一共有两个事务，一个是在Source到Channel之间，一个是Channel到Sink之间。在Source到Channel之间的叫put事务，在Channel到Sink之间的叫Take事务

• 从source到channel过程中，数据在flume中会被封装成Event对象，也就是一批event，把这批event放到一个事务中，把这个事务也就是这批event一次性的放入channel中。同理，Take事务的时候，也是把这一批event组成的事务统一拿出来到sink放到HDFS上

• Flume事务失败会进行回滚

9、Spark的容错机制

• RDD lineage血统图：

  • RDD的Lineage会记录RDD的元数据信息和转换行为，当该RDD的部分分区数据丢失时，它可以根据这些信息来重新运算和恢复丢失的数据分区（只支持粗粒度转换）

• Checkpoint机制：

  • 检查点（本质是通过将RDD写入Disk做检查点）是为了通过lineage做容错的辅助。lineage过长会造成容错成本过高

  • 计算此数据的花费时间较长并且对数据的持久的安全性有要求则使用checkpoint，宽依赖建议使用checkpoint

• spark的Broadcast：广播变量，就是不把副本变量分发到每个 Task 中，而是将其分发到每个 Executor，Executor 中的所有 Task 共享一个副本变量

10、Spark和MapReduce的区别

• Spark处理数据的设计模式与MR不同，Spark把中间的计算结果放在内存中，减少了磁盘IO，而MR在处理数据时涉及到多次磁盘IO操作

• Spark基于RDD的设计可以进行高效的迭代计算，而MR无法实现迭代式计算

• Spark基于DAG的血统图和checkpoint机制可以实现很好的容错性

• MR只有map和reduce两种操作，如果需要实现复杂的操作则需要进行多个MR的复合设计，而Spark丰富的算子操作可以简单的设计完成复杂的操作

11、Spark性能调优

• 在提交spark程序的时候合理的配置资源

• 对多次使用的RDD进行持久化或者Checkpoint

• 在reduceByKey和groupByKey中进行合理的选择

• 对shuffle阶段进行优化

• 广播共享数据

• 优化所使用的数据结构

• 使用序列化的持久化级别，将数据序列化之后再持久化，可以减小内存开销

12、YARN容错机制

• map阶段或者reduce阶段中代码异常：

  • jvm会向application master发送错误报告，并将此次任务标记为fail，当application master被告知一个任务尝试失败后，将重新调度该任务执行，application master会试图避免在以前失败过的节点管理器上重新调度该任务。此外如果一个任务失败过4次，将不会再重试运行任务的最多尝试次数：mapreduce.map.maxattempts设置)，默认情况下失败次数大于4次，整个作业会失败

• 任务JVM突然退出：

  • 在这种情况下，节点管理器会通知application master将此次任务尝试标记为失败。在此之后，任务JVM进程将被自动杀死。任务被认为失败的时间默认为10分钟

• application master运行失败：

  • application master向资源管理器发送周期性的心跳，当application master失败时，资源管理器将检测到该失败并在一个新的容器（由节点管理器管理）中开始一个新的master实例。对于MapReduce application master，它将使用作业历史来恢复失败的应用程序所运行任务的状态，使其不必重新运行。默认情况下恢复功能是开启的

• NodeManager运行失败：

  • 如果节点管理器由于崩溃或运行非常缓慢而失败，就会停止向资源管理器发送心跳信息（或发送频率很低）。如果10分钟内没有收到一条心跳信息，资源管理器将会通知停止发送心跳信息的节点管理器，并且将其从自己的节点池中移除以调度启用容器，在失败的节点管理器上运行的所有任务或application master都会在其他节点上进行恢复

• resourceManager运行失败：

  • 可以通过高可用配置两个资源管理器，如果主资源管理器失败了，那么备份资源管理器能够接替，且客户端不会感到明显的中断

13、spark中的partitioner（分区器）

• HashPartitioner：

  • 它使用Java的Object.hashCode实现基于散列的分区，hashcode()的概念是相等的对象应该具有相同的哈希码。 因此，基于此hashcode()概念，HashPartitioner将划分具有相同hashcode()的键到同一分区，HashPartitioner是Spark的默认分区器

• RangePartitioner：

  • 按范围将 可排序记录 划分为大致相等的范围。 范围通过对传入的RDD的内容进行采样来确定。即RangePartitioner将根据Key对记录进行排序，然后根据给定的值将记录划分为多个分区

• CustomPartitioner（自定义分区器）：

  • 我们还可以通过扩展Spark中的默认Partitioner类来自定义我们需要的分区数以及应该存储在这些分区中的内容。然后通过partitionBy()在RDD上应用自定义分区逻辑

• Spark从HDFS读入文件的分区数默认等于HDFS文件的块数(blocks)，HDFS中的block是分布式存储的最小单元。如果我们上传一个30GB的非压缩的文件到HDFS，HDFS默认的块容量大小128MB，因此该文件在HDFS上会被分为235块(30GB/128MB)；Spark读取SparkContext.textFile()读取该文件，默认分区数等于块数即235

14、spark yarn模式下的cluster模式和client模式的区别

• cluster模式：Driver程序在YARN中运行，应用的运行结果不能在客户端显示，所以最好运行那些将结果最终保存在外部存储介质（如HDFS、Redis、Mysql）而非stdout输出的应用程序，客户端的终端显示的仅是作为YARN的job的简单运行状况

• client模式：Driver运行在Client上，应用程序运行结果会在客户端显示，所有适合运行结果有输出的应用程序（如spark-shell）（master通过rpc与worker进行通信，通知worker启动一个或多个executor进程）

• Driver（applicationMaster）：负责整个应用任务的job的划分和stage的切割以及task的切割和优化，并负责把task分发到worker对应的节点的executor进程中的task线程中执行, 并获取task的执行结果，Driver通过SparkContext对象与spark集群获取联系，得到master主机host，就可以通过rpc向master注册自己

• Spark On Yarn 的优势 ：1. Spark 支持资源动态共享，运行于 Yarn 的框架都共享一个集中配置好的资源池 2. 可以很方便的利用 Yarn 的资源调度特性来做分类·，隔离以及优先级控制负载，拥有更灵活的调度策略 3. Yarn 可以自由地选择 executor 数量 4. Yarn 是唯一支持 Spark 安全的集群管理器，使用 Yarn，Spark 可以运行于 Kerberized Hadoop 之上，在它们进程之间进行安全认证

15、RDD、DataFrame、DataSet 是什么，以及区别

• RDD：弹性分布式数据集，Spark中最基础的数据抽象，特点是RDD只包含数据本身，没有数据结果

• DataFrame：分布式数据容器，除了数据本身，还记录了数据的结构信息，及Schema（列名、列字段类型）

• DataSet：不仅包含数据本身，记录了数据的结构信息schema，还包含了数据集的类型，也就是真正把数据集做成了一个java对象的形式，需要先创建一个样例类case class，把数据做成样例类的格式，每一列就是样例类里的属性

16、Hive中的分区表和分桶表

• 分区表：

  • 分区可以理解为分类，通过分类把不同类型的数据放到不同的目录下

  • 分类的标准就是分区字段，可以一个，也可以多个

  • 分区表的意义在于优化查询。查询时尽量利用分区字段。如果不使用分区字段，就会全部扫描

  • 分区表中需要指定分区字段,其实分区字段就是目录名称,但是我们查询的时候却可以把分区字段当成查询条件来使用

• 分桶表：

  • 分桶是相对分区进行更细粒度的划分。分桶将整个数据内容某列按照属性值的hash值进行区分，如要安装name属性分为3个桶，就是对name属性值的hash值对3取摸，按照取模结果对数据分桶。如取模结果为0的数据记录存放到一个文件，取模为1的数据存放到一个文件，取模为2的数据存放到一个文件

17、Kafka监控应该怎么做

• 一套完整的kafka监控，应该包括：

  • 消费者监控，主要是存活告警，消费滞后告警

  • 生产者监控，主要是存活告警，生产者消费上游数据能力告警

  • broker监控，主要是存活告警，流量告警，isr列表，topic异常告警，control变换告警

• kafka实现监控的三种方法：

  • 使用kafka自带的命令行工具kafka-consumer-groups.sh脚本查看

  • 使用kafka Consumer API编程

  • 使用kafka自带的JMX监控指标

18、SparkStreaming与Kafka集成问题

• sparkStreaming 创建的 RDD 的分区数和 kafka partitions分区数是一致的

• 如果Spark程序初始化时未设置核数，则会默认分区数与CPU核数相同

• 在kafka中：一个消费线程可以对应若干个分区，但同一时刻一个分区只能被具体某一个消费线程消费

• Spark程序可以通过Web UI对程序运行时间进行监控

19、SparkSql运行原理

• 进行 DataFrame/Dataset/SQL 编程

• 如果是有效的代码，即代码没有编译错误，Spark 会将其转换为一个逻辑计划

• Spark 将此逻辑计划转换为物理计划，同时进行代码优化

• Spark 然后在集群上执行这个物理计划 (基于 RDD 操作)

20、SparkSql与Hive的区别

• Hive是一种基于HDFS的数据仓库，并且提供了基于SQL模型的，针对存储了大数据的数据仓库，进行分布式交互查询的查询引擎

• SparkSQL并不能完全替代Hive，它替代的是Hive的查询引擎，SparkSQL由于其底层基于Spark自身的基于内存的特点，因此速度是Hive查询引擎的数倍以上，Spark本身是不提供存储的，所以不可能替代Hive作为数据仓库的这个功能

• SparkSQL相较于Hive的另外一个优点，是支持大量不同的数据源，包括hive、json、parquet、jdbc等

• SparkSQL由于身处Spark技术堆栈内，基于RDD来工作，因此可以与Spark的其他组件无缝整合使用，配合起来实现许多复杂的功能

21、HQL与SQL语句的区别

• Hive不支持等值连接

  • SQL中对两表内联可以写成： •select * from dual a,dual b where a.key = b.key; •Hive中应为 •select * from dual a join dual b on a.key = b.key

• hive支持嵌入mapreduce程序，来处理复杂的逻辑

• hive支持将转换后的数据直接写入不同的表，还能写入分区、hdfs和本地目录

• hive不支持事务

22、Flink的优势

• Flink 是基于事件驱动 (Event-driven) 的应用，能够同时支持流处理和批处理

• 基于内存的计算，能够保证高吞吐和低延迟，具有优越的性能表现

• 支持精确一次 (Exactly-once) 语意，能够完美地保证一致性和正确性

• 分层 API ，能够满足各个层次的开发需求

• 支持高可用配置，支持保存点机制，能够提供安全性和稳定性上的保证

• 多样化的部署方式，支持本地，远端，云端等多种部署方案

• 具有横向扩展架构，能够按照用户的需求进行动态扩容

• 活跃度极高的社区和完善的生态圈的支持

23、大数据中的海量小文件问题

• LOSF(lots of samll file)，大小在1MB以内的文件称为小文件，百万级数量及以上称为海量文件

• 衡量存储系统性能主要有两个关键指标，即IOPS和数据吞吐量（IOPS (Input/Output Per Second) 即每秒的输入输出量 ( 或读写次数 ) ，是衡量存储系统性能的主要指标之一）

• 海量小文件LOSF问题：

  • 元数据管理低效

  • 数据布局低效（数据块零散分配在磁盘的不同位置上）

  • IO流访问流程复杂

• 磁盘最适合顺序的大文件I/O读写模式，但非常不适合随机的小文件I/O读写模式

• 海量小文件由于数量巨大，而且通常需要进行共享和并发访问，目前通常采用分布式系统进行存储，包括分布式文件系统和分布式对象存储系统，每个存储节点底层采用磁盘文件系统进行管理

• 优化策略：

  • 按照减少数据访问时间的优化思路，采用更高性能的硬件来提高LOSF性能

  • 小文件合并存储（减少了大量元数据，简化了IO访问流程）

  • 元数据管理优化（对元数据进行精简）

• Hadoop中的小文件问题：

  • 在NameNode（内存）中保存的元数据每一条大约占用150byte，而hdfs中的小文件指的是其大小远远小于默认的block块的大小

  • 大量小文件对性能的影响：

    • 大量的小文件的存在会占用大量的NameNode内存，会影响hdfs的扩展能力

    • MapReduce时每一个block块会用一个Map来处理，而大量的会造成集群资源的过量消耗

  • 小文件的产生情况：

    • 实时流处理：SparkStreaming实时流处理后的数据直接存储至hdfs中，而由于SparkStreaming的微批处理模式和RDD的特性，会导致产生大量的小文件

    • Reduce的设置不合理导致产生大量的小文件

    • 数据本身的特点：在 HDFS 上存储大量的图片、短视频、短音频等文件，由于这些文件的特点，会导致产生大量的小文件

  • hdfs中小文件问题的解决方案：

    • Hadoop Archive（HRA-文件归档）：是一种特殊的归档格式，Hadoop Archive映射到文件系统目录，一个HAR是以扩展名 .har结尾 ，一个HAR目录包含元数据（以index和masterindex的形式）和data（part-*）文件，Hadoop Archive是一个高效地将小文件放入HDFS块中的文件存档工具，它能将多个小文件打包成一个HAR文件，这样在减少NameNode内存使用的同时，仍然允许对文件进行透明的访问

    • SequenceFile：是一个由二进制序列化过的key/value字节流组成的文本存储文件，它可以在map/reduce过程中的input/output的format时被使用。通过改变文件的写出方式，写入到SequenceFile格式的文件中（因为文件内容很小，可以以文件名作为key，文件内容作为value，直接写到SequenceFile中）

    • hive中的大量小文件：1.动态分区插入数据，产生大量的小文件，从而导致map数量剧增。 2.reduce数量越多，小文件也越多(reduce的个数和输出文件是对应的)。 3.数据源本身就包含大量的小文件 也可以使用SequenceFile来解决

• SparkStreaming中的小文件解决办法：

  • 增加batch的大小

  • 对SparkStreaming处理的数据进行再次合并处理后进行存储

大数据部分3

1、Hive中的数据倾斜

• 数据倾斜就是数据的分布不平衡

• key的分布不均匀或者说是某些key太集中

  • key的分布不均匀就会导致MR过程中的某些reduce的需要处理的数据量特别大，有些特别小，就会导致整个任务因为需要等待数据量特别大的reduce而迟迟无法完成

  • 解决办法：

    • 设置hive.map.aggr=true //开启map端部分聚合功能，就是将key相同的归到一起

    • 设置hive.groupby.skewindata=true //如果发生了数据倾斜就可以通过它来进行负载均衡

• 业务数据自身的特性

  • 某些业务数据作为key的字段本身就很集中，会导致数据倾斜（本质也是key分布不均匀）

• SQL语句造成的数据倾斜

  • 进行表的join这种业务操作时，经常会产生数据倾斜，因为数据要进入reduce，肯定要先进行分区，而分区就是根据key的hash值来进行的，既然数据的key本身就是不均匀的了（即某些key很集中，或者干脆就是有很多相同的key，比如key为无效值null），那这样分区的结果就是这些集中的key会被分到同一个分区中，而这些key的数量本就大，所以会产生数据倾斜

  • 解决办法：

    • 大小表Join： 使用map join让表（1000条以下的记录条数）先进内存，在Map端完成Reduce

    • 大表Join大表：

      • 当一张表中的大部分值都是0或者null时，容易shuffle给一个reduce，造成数据倾斜，而这种情况可以通过给异常key赋一个随机的值来分散key，来解决数据倾斜

      • 当key值都是有效值时，可以通过设定（set hive.optimize.skewjoin = true; set hive.skewjoin.key = skew_key_threshold （default = 100000））来控制倾斜的阈值，如果超过这个值，新的值会发送给那些还没有达到的reduce

• group by造成的数据倾斜：

  • 对数据做类型统计的时候，如果按类型进行group by则当某种类型的数据量特别多的时候，会出现某些reduce的处理数据量特别大，从而导致数据倾斜

  • 可以通过设置（hive.map.aggr=true（默认true） 这个配置项代表是否在map端进行聚合，相当于Combiner hive.groupby.skewindata=true（默认false），有数据倾斜的时候进行负载均衡，当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作

2、Hive中的分区、分桶、内部、外部表

• 内部表：未被external修饰的表为内部表，内部表数据由hive自身管理，在删除表的时候会删除其元数据以及存储数据

• 外部表：建表时被external修饰的为外部表，外部表的数据由hdfs管理，在删除表的时候只会删除元数据，而由hdfs管理的文件则不会被删除

• 分区表：分区可以理解为分类，通过分类把不同类型的数据放到不同的目录下，分区表中需要指定分区字段，然后可以在查询的时候将分区字段作为条件，查询指定的数据到指定的分区字段下，是对hive的一种优化

• 分桶表：分桶是相对分区更细致的划分，分桶按照分桶字段的hash值去模除以分桶的个数，如要按照name属性分为3个桶，就是对name属性值的hash值对3取摸，按照取模结果对数据分桶

• 分桶的作用：

  • 获得更高的查询处理效率，桶为表加上了额外的结构，Hive 在处理有些查询时能利用这个结构

  • 使取样（sampling）更高效，在处理大规模数据集时，在开发和修改查询的阶段，如果能在数据集的一小部分数据上试运行查询，会带来很多方便

  • 最大的作用是用来提高join操作的效率

• hive的动态分区：hive提供了一个动态分区功能，其可以基于查询参数的位置去推断分区的名称，从而建立分区，可以通过配置参数实现

  • set hive.exec.dynamic.partition =true（默认false）,表示开启动态分区功能；

  • set hive.exec.dynamic.partition.mode = nonstrict(默认strict),表示允许所有分区都是动态的，否则必须有静态分区字段

3、数仓分层

• ETL（数据仓库技术）是将业务系统的数据经过抽取、清洗转换之后加载到数据仓库的过程，目的是将企业中的分散、零乱、标准不统一的数据整合到一起，为企业的决策提供分析依据

• 数据仓库：是对整合的多个数据源的历史数据进行细粒度的、多维的分析，帮助高层管理者或者业务分析人员做出商业战略决策或商业报表

• 数据仓库分为四层：

  • ODS层（原始数据层）：存放原始数据，直接加载原始日志、数据，数据保存原貌不做处理

  • DWD层（明细数据层）：结构与粒度原始表保持一致，对ODS层数据进行清洗（去除空值、脏数据、超过极限范围的数据）

  • DWS（服务数据层）：以DWD为基础，进行轻度汇总

  • ADS（数据应用层）：为各种统计报表提供数据

• 为什么要分层：

  • 空间换时间：通过建设多层次的数据模型供用户使用，避免用户直接使用操作型数据，可以更高效的访问数据

  • 把复杂问题简单化：一个复杂的任务分解成多个步骤来完成，每一层只处理单一的步骤，比较简单和容易理解。而且便于维护数据的准确性，当数据出现问题之后，可以不用修复所有的数据，只需要从有问题的步骤开始修复

  • 便于处理业务的变化：随着业务的变化，只需要调整底层的数据，对应用层对业务的调整零感知

• 数据仓库与数据集市区别：

  • 数据仓库是企业级的，能为整个企业各个部门的运行提供决策支持手段

  • 数据集市则是一种微型的数据仓库,它通常有更少的数据,更少的主题区域,以及更少的历史数据,因此是部门级的，一般只能为某个局部范围内的管理人员服务，因此也称之为部门级数据仓库

4、数仓建模

• 数仓建模阶段划分：

  • 业务建模：生成业务模型，主要解决业务层面的分解和程序化

  • 领域概念建模：生成领域模型，主要是对业务模型进行抽象处理，生成领域概念模型

  • 逻辑建模：生成逻辑模型，主要是将领域模型的概念实体以及实体之间的关系进行数据库层次的逻辑化

  • 物理建模：生成物理模型，主要解决，逻辑模型针对不同关系型数据库的物理化以及性能等一些具体的技术问题

5、Spark垃圾回收

• 在Spark程序运行时Executor内存主要用于：1、用于缓存RDD（cache） 2、分配给task，在task执行过程中创建对象

• 默认情况下，Executor占用的内存，60%划分给RDD缓存，40%分配给task，存放运行期间创建的对象，这种默认情况可能会导致task用于运行期间创建对象的内存偏小了。当创建的对象把40%的内存空间用完，就会产生垃圾回收（GC），把不再被使用的对象清除出内存。所以说，如果给task分配的内存过低，会导致GC频繁发生，从而导致task工作线程频繁停止，降低了spark计算效率

• SparkConf().set("spark.storage.memoryFraction","比例")，此处的比例为RDD缓存的内存分配比例，rdd的持久化也可通过序列化来降低内存的占用

6、Hive的优化

• hive的小文件问题：

  • 动态分区插入数据，产生大量的小文件，从而导致map数量剧增（动态分区插入：通过在扫描输入表时动态确定应该创建和填充的分区）

  • reduce数量越多，小文件也越多

  • 数据源本身就包含大量的小文件

• 小文件问题的影响：

  • 从Hive的角度看，小文件会开很多map，一个map开一个JVM去执行，所以这些任务的初始化，启动，执行会浪费大量的资源，严重影响性能

  • 在HDFS中，每个小文件对象约占150byte，如果小文件过多会占用大量内存。这样NameNode内存容量严重制约了集群的扩展

• 小文件问题的解决：

  • 通过参数设置减少reduce的数量

  • 使用hadoop archive命令把小文件进行归档

  • 使用Sequencefile作为表存储格式在一定程度上可以减少小文件数量

• 设置合理的map/reduce数量

• 对SQL语句进行优化

• 通过设置分区表优化查询

7、SparkSql特点

• 能够将 SQL 查询与 Spark 程序无缝混合，允许您使用 SQL 或 DataFrame API 对结构化数据进行查询

• 支持多达上百种的外部数据源，包括 Hive，Avro，Parquet，ORC，JSON 和 JDBC 等

• 支持 HiveQL 语法以及 Hive SerDes 和 UDF，允许你访问现有的 Hive 仓库

• 支持标准的 JDBC 和 ODBC 连接

• 支持扩展并能保证容错

• SparkSql窗口函数：

  • 窗口函数是spark sql模块从1.4之后开始支持的，主要用于解决对一组数据进行操作，同时为每条数据返回单个结果，比如计算指定访问数据的均值、计算累进和或访问当前行之前行数据等

  • Spark SQL支持三类窗口函数：排名函数、分析函数和聚合函数

8、SparkStreaming与Flink的区别

• 处理模型与延迟：

  • Flink的处理模型为单条事件处理，亚秒级低延迟

  • SparkStreaming的处理模型为一个窗口内的所有事件，秒级延迟

• 任务调度：

  • Spark Streaming 是基于微批处理的，实际上每个批次都是一个 Spark Core 的任务。对于编码完成的 Spark Core 任务在生成到最终执行结束主要包括以下几个部分：

    • 构建 DGA 图；

    • 划分 stage；

    • 生成 taskset；

    • 调度 task

  • 对于 flink 的流任务客户端首先会生成 StreamGraph，接着生成 JobGraph，然后将 jobGraph 提交给 Jobmanager 由它完成 jobGraph 到 ExecutionGraph 的转变，最后由 jobManager 调度执行

• 运行角色不同：

  • Spark Streaming 运行时的角色(standalone 模式)主要有：

    • Master: 主要负责整体集群资源的管理和应用程序调度

    • Worker: 负责单个节点的资源管理，driver 和 executor 的启动等

    • Driver: 用户入口程序执行的地方，即 SparkContext 执行的地方，主要是 DGA 生成、stage 划分、task 生成及调度

    • Executor: 负责执行 task，反馈执行状态和执行结果

  • Flink 运行时的角色(standalone 模式)主要有:

    • Jobmanager: 协调分布式执行，他们调度任务、协调 checkpoints、协调故障恢复等。至少有一个 JobManager。高可用情况下可以启动多个 JobManager，其中一个选举为 leader，其余为 standby

    • Taskmanager: 负责执行具体的 tasks、缓存、交换数据流，至少有一个 TaskManager

    • Slot: 每个 task slot 代表 TaskManager 的一个固定部分资源，Slot 的个数代表着 taskmanager 可并行执行的 task 数

9、Spark on yarn 的提交流程

• Spark Yarn Client 向 Yarn 中提交应用程序

• ResourceManager 收到请求后，在集群中选择一个 NodeManager，并为该应用程序分配一个 Container，在这个 Container 中启动应用程序的 ApplicationMaster， ApplicationMaster 进行 SparkContext 等的初始化（一个Job对应一个AM）

• ApplicationMaster 向 ResourceManager 注册，这样用户可以直接通过 ResourceManager 查看应用程序的运行状态，然后采用轮询的方式通过RPC协议为各个任务申请资源，并监控它们的运行状态

• ApplicationMaster 申请到资源（也就是Container）后，Executor会向SparkContext注册并申请task

• Executor运行task并向ApplicationMaster汇报运行状态和进度

• 应用程序运行完成后，ApplicationMaster 向 ResourceManager申请注销并关闭自己

10、Yarn的架构

• ResourceManager

• ResourceManager 通常在独立的机器上以后台进程的形式运行，它是整个集群资源的主要协调者和管理者。ResourceManager 负责给用户提交的所有应用程序分配资源，它根据应用程序优先级、队列容量、ACLs、数据位置等信息，做出决策，然后以共享的、安全的、多租户的方式制定分配策略，调度集群资源

• NodeManager

• NodeManager 是 YARN 集群中的每个具体节点的管理者。主要负责该节点内所有容器的生命周期的管理，监视资源和跟踪节点健康

  • 启动时向 ResourceManager 注册并定时发送心跳消息，等待 ResourceManager 的指令

  • 维护 Container 的生命周期，监控 Container 的资源使用情况

  • 管理任务运行时的相关依赖，根据 ApplicationMaster 的需要，在启动 Container 之前将需要的程序及其依赖拷贝到本地

• ApplicationMaster

  • 在用户提交一个应用程序时，YARN 会启动一个轻量级的进程 ApplicationMaster。ApplicationMaster 负责协调来自 ResourceManager 的资源，并通过 NodeManager 监视容器内资源的使用情况，同时还负责任务的监控与容错

    • 根据应用的运行状态来决定动态计算资源需求

    • 向 ResourceManager 申请资源，监控申请的资源的使用情况

    • 跟踪任务状态和进度，报告资源的使用情况和应用的进度信息

    • 负责任务的容错

• YarnChild

  • 真正执行task的进程

• Container

  • Container 是 YARN 中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等。当 AM 向 RM 申请资源时，RM 为 AM 返回的资源是用 Container 表示的。YARN 会为每个任务分配一个 Container，该任务只能使用该 Container 中描述的资源

11、Kafka丢失数据的情况

• producer配置acks=0 ，1

• kafka的数据一开始就是存储在PageCache上的，定期flush到磁盘上的，也就是说，不是每个消息都被存储在磁盘了，如果出现断电或者机器故障等，PageCache上的数据就丢失了

• 网络负载很高或者磁盘很忙写入失败的情况下（无重发重试）

• 消费者崩溃，还没有处理的数据就被commit offset了

12、HDFS的block块相关

• 为什么hdfs中的block块设置的较大：

  • hdfs块比磁盘的块大，其目的是为了最小化寻址开销。如果块设置的足够大，那么从磁盘传输数据的时间会明显大于定位这个块开始位置所需的时间，因而，传输一个由多个块组成的文件时间取决于磁盘的传输速率

• hdfs中block为什么设置为128MB：

  • hdfs中的平均寻址时间为10ms，而传输速率普遍为100MB/s而在大量测试中得出寻址时间为传输时间的1%时，为最佳状态

  • 通过计算最佳传输时间=10ms/0.01=1s,进一步计算出Block块大小=1s*100MB=100MB,而我们磁盘block块的大小都是2^n倍，所以设置hdfs中的blcok大小为128MB

• 如果block设置过大：

  • 在MapReduce中的map任务通常一次只处理一个块中的数据，因此设置的过大就会使作业的运行速度较慢

• hdfs中小于blcok块大小的文件如何存储：

  • 当所存储的文件小于block块的大小时，它并不会占用整个block块的大小，只会占用它的实际存储大小

13、Hive的相关语法问题

• order by 会对输入做全局排序，因此只有一个reducer，会导致当输入规模较大时，需要较长的计算时间

• sort by不是全局排序，其在数据进入reducer前完成排序。因此，如果用sort by进行排序，并且设置mapred.reduce.tasks>1，则sort by只保证每个reducer的输出有序，不保证全局有序

• distribute by根据distribute by指定的内容将数据分到同一个reducer

• distribute by类似MR中partition（自定义分区），进行分区，结合sort by使用（distribute by的分区规则是根据分区字段的hash码与reduce的个数进行模除后，余数相同的分到一个区），Hive要求DISTRIBUTE BY语句要写在SORT BY语句之前

• Hive的Join：

  • Hive 不支持非等值的连接，因为非等值连接非常难转化到 map/reduce 任务

  • 只支持等值查询，也可以join多于两个表

  • join 时，每次 map/reduce 任务的逻辑：

    • reducer 会缓存 join 序列中除了最后一个表的所有表的记录，再通过最后一个表将结果序列化到文件系统。这一实现有助于在 reduce 端减少内存的使用量。实践中，应该把最大的那个表写在最后（否则会因为缓存浪费大量内存）

14、Hive的join过程

• select * from a inner join b on a.id=b.id 返回两张表中都有的信息

• select * from a left join b on a.id=b.id 以前面的表作为主表和其他表进行关联，返回的记录数和主表的记录数相同，关联不上的字段用NULL

• select * from a right join b on a.id=b.id 以后面的表为主表，和前面的表做关联，返回的记录数和主表一致，关联不上的字段为NULL

• select * from a inner join b on a.id=b.id where b.id is null 查询两个表中不一样的数据（hive自从2.2以后开始支持不等值连接）

• 如果不指定MapJoin或者不符合MapJoin的条件，那么Hive解析器会将Join操作转换成Common Join,即：在Reduce阶段完成join，整个过程包含Map、Shuffle、Reduce阶段

• Map阶段：

  • 读取源表的数据，Map输出时候以Join on条件中的列为key，如果Join有多个关联键，则以这些关联键的组合作为key，Map输出的value为join之后所关心的(select或者where中需要用到的)列；同时在value中还会包含表的Tag信息，用于标明此value对应哪个表

• Shuffle阶段：

  • 根据key的值进行hash,并将key/value按照hash值推送至不同的reduce中，这样确保两个表中相同的key位于同一个reduce中

• Reduce阶段：

  • 根据key的值完成join操作，期间通过Tag来识别不同表中的数据

• MapJoin:

  • MapJoin通常用于一个很小的表和一个大表进行join的场景

  • 在map 端进行join，其原理是broadcast join，即把小表作为一个完整的驱动表来进行join操作

  • 要连接的各个表里面的数据会分布在不同的Map中进行处理。即同一个Key对应的Value可能存在不同的Map中。这样就必须等到 Reduce中去连接。要使MapJoin能够顺利进行，那就必须满足这样的条件：除了一份表的数据分布在不同的Map中外，其他连接的表的数据必须在每 个Map中有完整的拷贝。MAPJION会把小表全部读入内存中，在map阶段直接拿另外一个表的数据和内存中表数据做匹配，由于在map是进行了join操作，省去了reduce运行的效率也会高很多

15、SparkStreaming与Kafka

• Spark Streaming 中提供了如下三种位置策略，用于指定 Kafka 主题分区与 Spark 执行程序 Executors 之间的分配关系：

  • PreferConsistent : 它将在所有的 Executors 上均匀分配分区

  • PreferBrokers : 当 Spark 的 Executor 与 Kafka Broker 在同一机器上时可以选择该选项，它优先将该 Broker 上的首领分区分配给该机器上的 Executor

  • PreferFixed : 可以指定主题分区与特定主机的映射关系，显示地将分区分配到特定的主机，其构造器如下：

    • @Experimental def PreferFixed(hostMap: collection.Map[TopicPartition, String]): LocationStrategy =  new PreferFixed(new ju.HashMap[TopicPartition, String](hostMap.asJava))  @Experimental def PreferFixed(hostMap: ju.Map[TopicPartition, String]): LocationStrategy =  new PreferFixed(hostMap)

• SparkStreaming消费Kafka的方式：

  • 基于Receiver的方式：把数据从kafka中读取出来然后缓存到内存然后再定时处理（会产生数据丢失的风险 如果要保证高可用必须开启WAL机制，影响性能）

  • 基于Direct的方式：周期性地查询kafka，来获得每个topic+partition的最新的offset，并且主动的进行数据获取

    • 可以简化并行读取：spark会创建跟kafka partition一样多的RDD partition，并且会并行从kafka中读取数据

    • 高性能：kafka中做了数据复制，可以通过kafka的副本进行恢复

    • 缺点是成本提高且无法通过zookeeper来监控消费者消费情况

16、Hadoop相关

• NN与DN之间的交互：DataNode 会有一个后台线程，定时向NameNode 发送每个块的最新状态及信息，通过RPC通信

• HA的实现：

  • 创建JournalNode集群来存储各NameNode所需的元数据信息，从而保持元数据信息同步

  • 每个NameNode都有一个对应的ZKFC作为自己的监护线程，HDFS系统初次启动时，多个NameNode对应的ZKFC会进行锁的争夺（即Zookeeper中的节点注册），抢到锁的ZKFC会将对应的NameNode置成Active。若是ZKFC在监视过程发现自己负责的Active NameNode没有信息反馈，则认为该NameNode已经故障，则会将抢到的锁释放。其他的ZKFC会对释放的锁进行争夺，获得者会进入原Active NameNode中将原Active NameNode 线程杀掉，然后将自己对应的NameNode置为Active对外服务，从而解决了单NameNode故障，集群无法工作及Active NameNode的选择问题

17、Spark累加器与广播变量

• 在 Spark 中，提供了两种类型的共享变量：累加器 (accumulator) 与广播变量 (broadcast variable)：

  • 累加器：用来对信息进行聚合，主要用于累计计数等场景

  • 广播变量：主要用于在节点间高效分发大对象

• 累加器（accumulator）是 Spark 中提供的一种分布式的变量机制，其原理类似于 mapreduce，即分布式的改变，然后聚合这些改变。累加器的一个常见用途是在调试时对作业执行过程中的事件进行计数

18、Kafka消费者

• 为什么要有消费者组：（一个分区只可以被一个消费者组中的消费者消费）

  • 消费者通常是消费者群组的一部分，多个消费者群组共同读取同一个主题时，彼此之间互不影响

  • 之所以引进消费者组是因为，Kafka 消费者经常会做一些高延迟的操作，比如把数据写到数据库或 HDFS ，或者进行耗时的计算，在这种情况下单个消费者无法跟上数据生成的速度，则此时可以增加更多的消费者，让它们分担负载，分别处理部分分区的消息，这就是 Kafka 实现横向伸缩的主要手段

  • 所以应该合适的设置消费者的数量，以免造成闲置和额外开销

• 分区再均衡：

  • 因为群组里的消费者共同读取主题的分区，所以当一个消费者被关闭或发生崩溃时，它就离开了群组，原本由它读取的分区将由群组里的其他消费者来读取

  • 同时在主题发生变化时 ， 比如添加了新的分区，也会发生分区与消费者的重新分配，分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为再均衡。正是因为再均衡，所以消费费者群组才能保证高可用性和伸缩性

  • 消费者通过向群组协调器所在的 broker 发送心跳来维持它们和群组的从属关系以及它们对分区的所有权。只要消费者以正常的时间间隔发送心跳，就被认为是活跃的，说明它还在读取分区里的消息。消费者会在轮询消息或提交偏移量时发送心跳。如果消费者停止发送心跳的时间足够长，会话就会过期，群组协调器认为它已经死亡，就会触发再均衡

• 偏移量：

  • Kafka 的每一条消息都有一个偏移量属性，记录了其在分区中的位置，偏移量是一个单调递增的整数

  • 如果不能正确的提交偏移量则会导致数据丢失或者重复消费

  • Kafka 支持自动提交和手动提交偏移量两种方式

• 监听分区再均衡：因为分区再均衡会导致分区与消费者的重新划分，有时候你可能希望在再均衡前执行一些操作：比如提交已经处理但是尚未提交的偏移量，关闭数据库连接等。此时可以在订阅主题时候，调用 subscribe 的重载方法传入自定义的分区再均衡监听器

• Kafka 的设计目标是高吞吐和低延迟，所以在 Kafka 中，消费者通常都是从属于某个群组的，这是因为单个消费者的处理能力是有限的（也可以设置单个消费者，只需要把主题或者分区分配给消费者，然后开始读取消息井提交偏移量即可）

19、Kafka生产者

• 生产者发送消息的过程：

  • Kafka 会将发送消息包装为 ProducerRecord 对象， ProducerRecord 对象包含了目标主题和要发送的内容，同时还可以指定键和分区。在发送 ProducerRecord 对象前，生产者会先把键和值对象序列化成字节数组，这样它们才能够在网络上传输

  • 接下来，数据被传给分区器。如果之前已经在 ProducerRecord 对象里指定了分区，那么分区器就不会再做任何事情。如果没有指定分区 ，那么分区器会根据 ProducerRecord 对象的键来选择一个分区，紧接着，这条记录被添加到一个记录批次里，这个批次里的所有消息会被发送到相同的主题和分区上。有一个独立的线程负责把这些记录批次发送到相应的 broker 上

  • 服务器在收到这些消息时会返回一个响应。如果消息成功写入 Kafka，就返回一个 RecordMetaData 对象，它包含了主题和分区信息，以及记录在分区里的偏移量。如果写入失败，则会返回一个错误。生产者在收到错误之后会尝试重新发送消息，如果达到指定的重试次数后还没有成功，则直接抛出异常，不再重试

• Kafka 有着默认的分区机制：

  • 如果键值为 null， 则使用轮询 (Round Robin) 算法将消息均衡地分布到各个分区上

  • 如果键值不为 null，那么 Kafka 会使用内置的散列算法对键进行散列，然后分布到各个分区上

• 可以通过实现Partitioner来自定义分区器，然后再创建生产者时指定分区器和所需要配置的参数

20、kafka副本机制

• 为了保证高可用，kafka 的分区是多副本的，如果一个副本丢失了，那么还可以从其他副本中获取分区数据

• 分区：kafka的主题被分为多个分区，分区是kafka最基本的存储单位，每个分区都可以有多个副本，其中一个副本是首领副本 (Leader replica)，所有的事件都直接发送给首领副本；其他副本是跟随者副本 (Follower replica)，需要通过复制来保持与首领副本数据一致，当首领副本不可用时，其中一个跟随者副本将成为新首领

• ISR机制：每个分区都有5一个 ISR(in-sync Replica) 列表，用于维护所有同步的、可用的副本

• 同步副本：

  • 与 Zookeeper 之间有一个活跃的会话，即必须定时向 Zookeeper 发送心跳

  • 在规定的时间内从首领副本那里低延迟地获取过消息

21、Yarn的作用

• hadoop1.x版本的mapreduce执行任务时主要的两种后台服务进程：

  • JobTracker：负责资源管理，负责接收客户作业提交，调度任务到工作节点上运行，并提供诸如监控工作节点状态及任务进度等管理功能

  • TaskTracker：通过周期性的心跳来通知jobtracker其当前的健康状态，每一次心跳包含了可用的map和reduce任务数目、占用的数目以及运行中的任务详细信息

  • 该版本mapreduce的缺点：

    • JobTracker容易存在单点故障

    • jobTracker负担重，既要负责资源管理，又要进行作业调度；当需处理太多任务时，会造成过多的资源消耗

    • 在TaskTracker端，把资源强制划分为map task slot和reduce task slot，如果当系统中只有map task或者只有reduce task的时候，会造成资源的浪费

• hadoop2.x上的mapreduce是基于YARN 的，YARN支持多个计算框架

• Yarn的作用：

  • YARN的设计减小了JobTracker的资源消耗

  • 对于资源的表示以内存为单位，比之前的表示方式更加合理

  • MRv1中JobTracker一个很大的负担就是监控job下的tasks的运行状况，现在这个部分由ApplicationMaster来实现

  • Container用来作为YARN的一个资源隔离组件，可以用来对资源进行调度和控制

22、MapReduce

• 在shuffle过程中ruduce如何去对应的maptask拉去数据：

  • 每个reduce task不断地通过RPC从ApplicationMaster那里获取map task是否完成的信息（还有存储位置等），如果reduce task得到通知，获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程开始启动

• mapreduce中的两次排序：

  • 第一次排序发生在map阶段，在map阶段环形（内存）缓冲区的数据写入到磁盘的时候会根据reducetask的数量生成对应的分区，然后根据对应数据的哈希对分区数取模写入，然后会根据key值对分区中的数据使用快速排序算法进行排序，所以每个分区内的数据是有序的

  • 第二次排序发生在reduce阶段，reducetask去每一个maptask所在的节点上的对应的分区（可以根据分区好）上拉取数据，然后因为分区写入使用哈希取模算法，所以相同的数据必定写在同一reduce分区中，采用归并排序对所拉取数据的key进行排序

  • 最后有几个reduce分区就会生成几个partition文件

23、ZooKeeper相关

• zookeeper的选举Master机制：通过过半选举机制产生leader（投票选举机制 ID大的优先获得投票）

• zookeeper集群中节点数设置：

  • zookeeper选举规则：leader选举，要求 可用节点数量 > 总节点数量/2 （为了防止在集群出现脑裂的时候，可能会出现多个子集群同时服务的情况（即子集群各组选举出自己的leader）， 这样对整个zookeeper集群来说是紊乱的）

  • 集群脑裂：集群的脑裂通常是发生在节点之间通信不可达的情况下，集群会分裂成不同的小集群，小集群各自选出自己的master节点，导致原有的集群出现多个master节点的情况，这就是脑裂

  • 所以可以通过leader选举规则，在容错能力相同的条件下，奇数个节点更加节省资源

• zookeeper的数据模型：Zookeeper 通过树形结构来存储数据，它由一系列被称为 ZNode 的数据节点组成，类似于常见的文件系统。不过和常见的文件系统不同，Zookeeper 将数据全量存储在内存中，以此来实现高吞吐，减少访问延迟

• zookeeper集群：可以由一组 Zookeeper 服务构成 Zookeeper 集群，集群中每台机器都会单独在内存中维护自身的状态，并且每台机器之间都保持着通讯，只要集群中有半数机器能够正常工作，那么整个集群就可以正常提供服务

• 集群角色：

  • Leader ：为客户端提供读写服务，并维护集群状态，它是由集群选举所产生的

  • Follower ：为客户端提供读写服务，并定期向 Leader 汇报自己的节点状态。同时也参与写操作“过半写成功”的策略和 Leader 的选举

  • Observer ：为客户端提供读写服务，并定期向 Leader 汇报自己的节点状态，但不参与写操作“过半写成功”的策略和 Leader 的选举，因此 Observer 可以在不影响写性能的情况下提升集群的读性能

• 顺序访问：对于来自客户端的每个更新请求，Zookeeper 都会分配一个全局唯一的递增 ID，这个 ID 反映了所有事务请求的先后顺序

• 高性能：ZooKeeper 将数据存全量储在内存中以保持高性能，并通过服务集群来实现高可用，由于 Zookeeper 的所有更新和删除都是基于事务的，所以其在读多写少的应用场景中有着很高的性能表现

24、Hbase列族数量

25、Hive与关系型数据库比较

• Hive使用HDFS（hadoop的分布式文件系统），关系型数据库则是服务器本地的文件系统

• Hive使用的计算模型是MapReduce，而关系型数据库则是自己设计的计算模型

• Hive是为海量数据做数据挖掘分析而设计的，实时性差；而关系型数据库是为实时查询的业务进行设计的

• Hive是基于Hadoop的，所以Hive的可扩展性高

• Hive所支持的数据规模远远大于关系型数据库

• Hive是针对数据仓库而设计的，所以Hive的数据更新延迟高（且行级更新，插入需要通过配置文件实现）

26、Hive调优

• 根据使用场景的不同选择合适的文件存储格式

• MapReduce任务的主要瓶颈在网络IO和磁盘IO，所以应该选择合适的数据压缩方式来减少数据量

• 对Sql语句的优化

• 避免产生数据倾斜

27、Hive常用的存储格式

• TEXTFILE：按行存储，不支持块压缩，默认格式，数据不做压缩，磁盘开销大，加载数据的速度最高

• RCFILE：

  • 数据按行分块，每块按列存储，结合了行存储和列存储的优点

  • RCFile 保证同一行的数据位于同一节点，因此元组重构的开销很低

  • RCFile 能够利用列维度的数据压缩，并且能跳过不必要的列读取

• ORCFile：

  • 存储方式：数据按行分块 每块按照列存储

  • 压缩快 快速列存取

  • 效率比rcfile高,是rcfile的改良版本 使用了索引

  • 使用ORC文件格式可以提高hive读、写和处理数据的能力

• PARQUET：按列存储，相对于ORC，Parquet压缩比较低，查询效率较低

• SequenceFile：

  • Hadoop API提供的一种二进制文件，以<key,value>的形式序列化到文件中

  • 存储方式：行存储

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

• 如果是 Kafka 消费能力不足，则可以考虑增加 Topic 的分区数，并且同时提升消费 组的消费者数量，消费者数=分区数

• 如果是 Kafka 消费能力不足，则可以考虑增加 Topic 的分区数，并且同时提升消费 组的消费者数量，消费者数=分区数