大数据基础面试题二：Zookeeper&Flume&Kafka

红太狼 2022-11-05 08:50 281阅读 0赞

# 大数据基础面试题二：Zookeeper&Flume&Kafka #

### 目录 ###

*  大数据基础面试题二：Zookeeper&Flume&Kafka
 *   *  三、Zookeeper
     *   *  3.1 选举机制
         *  3.2 常用命令
         *  3.3 Paxos算法（扩展）
         *  3.4 讲一讲什么是CAP法则？Zookeeper符合了这个法则的哪两个？（扩展）
     *  四、Flume
     *   *  4.1 Flume组成，Put事务，Take事务
         *  4.2 Flume拦截器
         *  4.3 Flume Channel选择器
         *  4.4 Flume监控器
         *  4.5 Flume采集数据会丢失吗?（防止数据丢失的机制）
     *  五、Kafka
     *   *  5.1 Kafka架构
         *  5.2 Kafka的机器数量
         *  5.3 副本数设定
         *  5.4 Kafka压测
         *  5.5 Kafka日志保存时间
         *  5.6 Kafka中数据量计算
         *  5.7 Kafka的硬盘大小
         *  5.8 Kafka监控
         *  5.9 Kakfa分区数
         *  5.10 多少个Topic
         *  5.11 Kafka的ISR副本同步队列
         *  5.12 Kafka分区分配策略
         *  5.13 Kafka挂掉
         *  5.14 Kafka丢不丢数据
         *  5.15 Kafka数据重复
         *  5.16 Kafka消息数据积压，Kafka消费能力不足怎么处理？
         *  5.17 Kafka参数优化
         *  5.18 Kafka高效读写数据
         *  5.19 Kafka单条日志传输大小
         *  5.20 Kafka过期数据清理
         *  5.21 Kafka可以按照时间消费数据
         *  5.22 Kafka消费者角度考虑是拉取数据还是推送数据
         *  5.23 Kafka中的数据是有序的吗

## 三、Zookeeper ##

### 3.1 选举机制 ###

半数机制：2n+1，安装奇数台

> 10台服务器：3台  
> 20台服务器：5台  
> 100台服务器：11台  
> 台数多，好处：提高可靠性；坏处：影响通信延时

### 3.2 常用命令 ###

> ls、get、create、delete

### 3.3 Paxos算法（扩展） ###

Paxos算法一种基于消息传递且具有高度容错特性的一致性算法。

> 分布式系统中的节点通信存在两种模型：**共享内存**（Shared memory）和**消息传递**（Messages passing）。基于消息传递通信模型的分布式系统，不可避免的会发生以下错误：进程可能会慢、被杀死或者重启，消息可能会延迟、丢失、重复，在基础Paxos场景中，先不考虑可能出现消息篡改即拜占庭错误的情况。Paxos算法解决的问题是在一个可能发生上述异常的分布式系统中如何就某个值达成一致，保证不论发生以上任何异常，都不会破坏决议的一致性。

### 3.4 讲一讲什么是CAP法则？Zookeeper符合了这个法则的哪两个？（扩展） ###

CAP法则：强一致性、高可用性、分区容错性；

Zookeeper符合强一致性、高可用性！

## 四、Flume ##

### 4.1 Flume组成，Put事务，Take事务 ###

**1）taildir source**  
（1）断点续传、多目录  
（2）哪个Flume版本产生的？Apache1.7、CDH1.6  
（3）没有断点续传功能时怎么做的？ 自定义  
（4）taildir挂了怎么办？ 不会丢数：断点续传  
（5）怎么处理重复数据？

> 不处理：生产环境通常不处理，出现重复的概率比较低。处理会影响传输效率。  
> 处理  
> 自身：在taildirsource里面增加自定义事务，影响效率  
> 找兄弟：下一级处理（hive dwd sparkstreaming flink布隆）、去重手段（groupby、开窗取窗口第一条、redis）

（6）taildir source 是否支持递归遍历文件夹读取文件？

> 不支持。 自定义 递归遍历文件夹 + 读取文件

**2）file channel /memory channel/kafka channel**

（1）File Channel

> 数据存储于磁盘，优势：可靠性高；劣势：传输速度低  
> 默认容量：100万event  
> 注意：FileChannel可以通过配置dataDirs指向多个路径，每个路径对应不同的硬盘，增大Flume吞吐量。

（2）Memory Channel

> 数据存储于内存，优势：传输速度快；劣势：可靠性差  
> 默认容量：100个event

（3）Kafka Channel

> 数据存储于Kafka，基于磁盘；  
> 优势：可靠性高；  
> 传输速度快 Kafka Channel 大于Memory Channel + Kafka Sink 原因省去了Sink阶段

（4）Kafka Channel哪个版本产生的？

> Flume1.6 版本产生=》并没有火；因为有bug  
> topic-start 数据内容  
> topic-event 数据内容 ture 和false 很遗憾，都不起作用。  
> 增加了额外清洗的工作量。  
> Flume1.7解决了这个问题，开始火了。

（5）生产环境如何选择

> 如果下一级是Kafka，优先选择Kafka Channel  
> 如果是金融、对钱要求准确的公司，选择File Channel  
> 如果就是普通的日志，通常可以选择Memory Channel  
> 每天丢几百万数据 pb级 亿万富翁，掉1块钱会捡？

**3）HDFS sink**

（1）时间（1小时-2小时） or 大小128m、event个数（0禁止）

> 具体参数：hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount =0

**4）事务**

> Source到Channel是Put事务  
> Channel到Sink是Take事务

### 4.2 Flume拦截器 ###

**1）拦截器注意事项**

项目中自定义了：ETL拦截器。  
采用两个拦截器的优缺点：优点，模块化开发和可移植性；缺点，性能会低一些

**2）自定义拦截器步骤**  
（1）实现 Interceptor  
（2）重写四个方法

*  initialize 初始化
 *  public Event intercept(Event event) 处理单个Event
 *  public List intercept(List events) 处理多个Event，在这个方法中调用Event intercept(Event event)
 *  close方法

（3）静态内部类，实现Interceptor.Builder

**3）拦截器可以不用吗？**

> 可以不用；需要在下一级hive的dwd层和sparksteaming里面处理  
> 优势：只处理一次，轻度处理；劣势：影响性能，不适合做实时推荐这种对实时要求比较高的场景。

### 4.3 Flume Channel选择器 ###

Replicating：默认选择器。功能：将数据发往下一级所有通道  
Multiplexing：选择性发往指定通道。

### 4.4 Flume监控器 ###

1）采用Ganglia监控器，监控到Flume尝试提交的次数远远大于最终成功的次数，说明Flume运行比较差。  
2）解决办法？

> 1）自身：增加内存flume-env.sh 4-6g  
> \-Xmx与-Xms最好设置一致，减少内存抖动带来的性能影响，如果设置不一致容易导致频繁fullgc。  
> （2）找朋友：增加服务器台数  
> 搞活动 618 =》增加服务器=》用完在退出  
> 日志服务器配置：8-16g内存、磁盘8T

### 4.5 Flume采集数据会丢失吗?（防止数据丢失的机制） ###

如果是FileChannel不会，Channel存储可以存储在File中，数据传输自身有事务。  
如果是MemoryChannel有可能丢。

## 五、Kafka ##

### 5.1 Kafka架构 ###

生产者、Broker、消费者、ZK；  
注意：Zookeeper中保存Broker id和消费者offsets等信息，但是没有生产者信息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]

### 5.2 Kafka的机器数量 ###

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

### 5.3 副本数设定 ###

一般我们设置成2个或3个，很多企业设置为2个。  
副本的优势：提高可靠性；副本劣势：增加了网络IO传输

### 5.4 Kafka压测 ###

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

### 5.5 Kafka日志保存时间 ###

默认保存7天；生产环境建议3天

### 5.6 Kafka中数据量计算 ###

> 每天总数据量100g，每天产生1亿条日志， 10000万/24/60/60=1150条/每秒钟  
> 平均每秒钟：1150条  
> 低谷每秒钟：50条  
> 高峰每秒钟：1150条\*（2-20倍）=2300条-23000条  
> 每条日志大小：0.5k-2k（取1k）  
> 每秒多少数据量：2.0M - 20MB

### 5.7 Kafka的硬盘大小 ###

每天的数据量100g \* 2个副本 \* 3天 / 70%

### 5.8 Kafka监控 ###

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

### 5.9 Kakfa分区数 ###

1）创建一个只有1个分区的topic  
2）测试这个topic的producer吞吐量和consumer吞吐量。  
3）假设他们的值分别是Tp和Tc，单位可以是MB/s。  
4）然后假设总的目标吞吐量是Tt，那么分区数=Tt / min（Tp，Tc）

> 例如：producer吞吐量=20m/s；consumer吞吐量=50m/s，期望吞吐量100m/s；  
> 分区数=100 / 20 = 5分区  
> https://blog.csdn.net/weixin\_42641909/article/details/89294698  
> 分区数一般设置为：3-10个

### 5.10 多少个Topic ###

通常情况：多少个日志类型就多少个Topic。也有对日志类型进行合并的。

### 5.11 Kafka的ISR副本同步队列 ###

> ISR（In-Sync Replicas），副本同步队列。ISR中包括Leader和Follower。如果Leader进程挂掉，会在ISR队列中选择一个服务作为新的Leader。有replica.lag.max.messages（延迟条数）和replica.lag.time.max.ms（延迟时间）两个参数决定一台服务是否可以加入ISR副本队列，在0.10版本移除了replica.lag.max.messages参数，防止服务频繁的进去队列。  
> 任意一个维度超过阈值都会把Follower剔除出ISR，存入OSR（Outof-Sync Replicas）列表，新加入的Follower也会先存放在OSR中。

### 5.12 Kafka分区分配策略 ###

在 Kafka内部存在两种默认的分区分配策略：Range和 RoundRobin。

> 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进行排序，最后按照轮询的方式发给每一个消费线程。

### 5.13 Kafka挂掉 ###

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

### 5.14 Kafka丢不丢数据 ###

Ack = 0，相当于异步发送，消息发送完毕即offset增加，继续生产。  
Ack = 1，leader收到leader replica 对一个消息的接受ack才增加offset，然后继续生产。  
Ack = -1，leader收到所有replica 对一个消息的接受ack才增加offset，然后继续生产。

### 5.15 Kafka数据重复 ###

幂等性 + ack-1 + 事务  
Kafka数据重复，可以再下一级：SparkStreaming、redis或者Hive中dwd层去重，去重的手段：分组、按照id开窗只取第一个值；

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

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

### 5.17 Kafka参数优化 ###

1）Broker参数配置（server.properties）

1、日志保留策略配置
    # 保留三天，也可以更短 （log.cleaner.delete.retention.ms）
    log.retention.hours=72

2、Replica相关配置
    default.replication.factor:1 默认副本1个

3、网络通信延时
    replica.socket.timeout.ms:30000 #当集群之间网络不稳定时,调大该参数
    replica.lag.time.max.ms= 600000# 如果网络不好,或者kafka集群压力较大,会出现副本丢失,然后会频繁复制副本,导致集群压力更大,此时可以调大该参数

2）Producer优化（producer.properties）

compression.type:none                 gzip  snappy  lz4  
    #默认发送不进行压缩，推荐配置一种适合的压缩算法，可以大幅度的减缓网络压力和Broker的存储压力。

3）Kafka内存调整（kafka-server-start.sh）  
默认内存1个G，生产环境尽量不要超过6个G。

export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"

### 5.18 Kafka高效读写数据 ###

1）Kafka本身是分布式集群，同时采用分区技术，并发度高。  
2）顺序写磁盘  
Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。  
3）零复制技术  
![在这里插入图片描述][20210304231610360.png]

### 5.19 Kafka单条日志传输大小 ###

Kafka对于消息体的大小默认为单条最大值是1M但是在我们应用场景中，常常会出现一条消息大于1M，如果不对Kafka进行配置。则会出现生产者无法将消息推送到Kafka或消费者无法去消费Kafka里面的数据，这时我们就要对Kafka进行以下配置：server.properties

replica.fetch.max.bytes: 1048576  broker可复制的消息的最大字节数, 默认为1M
    message.max.bytes: 1000012   kafka 会接收单个消息size的最大限制， 默认为1M左右

注意：message.max.bytes必须小于等于replica.fetch.max.bytes，否则就会导致replica之间数据同步失败。

### 5.20 Kafka过期数据清理 ###

保证数据没有被引用（没人消费他）

日志清理保存的策略只有**delete**和**compact**两种  
log.cleanup.policy = delete启用删除策略  
log.cleanup.policy = compact启用压缩策略  
https://www.jianshu.com/p/fa6adeae8eb5

### 5.21 Kafka可以按照时间消费数据 ###

Map<TopicPartition, OffsetAndTimestamp> startOffsetMap = KafkaUtil.fetchOffsetsWithTimestamp(topic, sTime, kafkaProp);

### 5.22 Kafka消费者角度考虑是拉取数据还是推送数据 ###

拉取数据

### 5.23 Kafka中的数据是有序的吗 ###

单分区内有序；多分区，分区与分区间无序；  
扩展：  
kafka producer发送消息的时候，可以指定key:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]

这个key的作用是为消息选择存储分区，key可以为空，当指定key且不为空的时候，Kafka是根据key的hash值与分区数取模来决定数据存储到那个分区。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]

有序解决方案：同一张表的数据 放到 同一个 分区  
=> ProducerRecord里传入key，会根据key取hash算出分区号  
=> key使用表名，如果有库名，拼接上库名

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70]: /images/20221023/e389bd15d6914b568ca76bdcf5fffda6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 1]: /images/20221023/328a362e41304c52963112b1aa4d34ae.png
[20210304231610360.png]: /images/20221023/5ca1c232f0f040ecb3fd300ff194c311.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 2]: /images/20221023/8ad604bdd1d34964a7756c37cdd90458.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80Mjc5NjQwMw_size_16_color_FFFFFF_t_70 3]: /images/20221023/0156bb818ce3479d96588906bc2d42c8.png