关于Zookeeper的总结
1.3.1选举机制
半数机制： 2n 1； 10台服务器： 3台； 20台服务器： 5台； 100台服务器： 11台
台数并不是越多越好。 选举时间太长，影响性能。
1.3.2常用命令
ls、get、create
1.4流总结
1.4.1流配置、上传事务、Take事务
Taildir Source :断点之后的多目录。 在Flume1.6之前，每次读取文件时都必须自定义Source记录，以实现断点的重新分发。
文件通道：数据存储在磁盘上，并且可以存储宕机数据。 但是，传输速度很慢。 适用于金融业等要求数据传输可靠性的场景。
内存通道：数据存储在内存中，停机数据丢失。 传输很快。 适用于数据传输不可靠的场景，如常规日志数据。
Kafka channel :减少了flume的Sink阶段，提高了传输效率。
从Source到Channel是推送事务
从Channel到Sink是Take事务
1.4.2流拦截器
(1)窃听器注意事项
在中定制了。 ETL拦截器和分区拦截器。
采用两个拦截器的优缺点：优缺点、模块化开发和可移植性； 缺点、性能稍微变低
)2)定制拦截器的步骤
a )实现拦截器
b )改写四种方法
初始化初始化
公共事件接口(事件事件)处理单个事件
publiclisteventintercept [ listeventevents ]处理多个事件，该方法调用event intercept (事件事件事件)
关闭方法
c )静态内部类，Interceptor.Builder实现
1.4.3流通道选择器
1.4.4流监视器
安吉拉
1.4.5 Flume收集数据是否会丢失？ (防止数据丢失的机制)
不，通道存储可以保存在文件中。 数据传输本身就有事务。
1.4.6闪存
在开发中，在flume-env.sh中将JVM heap设置为4G或更高，并将其部署到其他服务器(4核8线程16G内存)
希望将-Xmx和-Xms设置为匹配，以减少内存抖动对性能的影响。 不一致容易导致频繁的全GC。
-Xms表示JVM堆内存(heap )的最小大小，初始分配； -Xmx表示JVM堆内存(heap )的最大允许大小，并根据需要进行分配。 如果不设置匹配，初始化时内存不足，因此容易频繁启动fullgc。
1.4.7文件通道优化
将dataDirs配置为指向多个路径，每个路径支持不同的硬盘，从而提高流吞吐量。
官方说明如下
commaseparatedlistofdirectoriesforstoringlogfiles.usingmultipledirectoriesonseparatediskscanimprovefilechannnelpeformance
checkpointDir和backupCheckpointDir也尽可能位于与不同硬盘相对应的目录中，以便在checkpoint损坏后，可以快速使用backupCheckpointDir恢复数据
1.4.8流通道容量
文件通道容量1000000个
内存通道容量100个
1.4.9 HDFS Sink小文件处理
)1) HDFS存储在大量小文件中有什么影响？
元数据级别：每个小文件都有元数据，如文件路径、文件名、所有者、所属组、权限和创建时间，这些信息存储在Namenode内存中。 因此，过多的小文件会占用Namenode服务的大量内存，影响Namenode的性能和寿命
计算级别：默认情况下，MR为每个小文件启用一个Map任务计算，这将严重影响计算性能。 也会影响磁盘地址时间。
(2) HDFS小文件处理
公式的默认三个参数配置在写入HDFS时会生成小文件。 hdfs.rollInterval、hdfs.rollSize、hdfs.rollCount
上述hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount=0，hdfs.roundValue=3600，HDFS.roundund unt
)1) tmp文件达到128M时滚动生成正式文件
)2) tmp文件生成时间超过3600秒时，滚动生成正式文件
例如，在2018-01-01 05:23时，sink接收数据并生成与以下示例类似的tmp文件：
/atguigu/
20180101/atguigu.201801010520.tmp即使文件内容没有达到128M，也会在06:23时滚动生成正式文件
1.5 Kafka相关总结
1.5.1 Kafka架构
1.5.2 Kafka压测
Kafka官方自带压力测试脚本（kafka-consumer-perf-test.sh、kafka-producer-perf-test.sh）。Kafka压测时，可以查看到哪个地方出现了瓶颈（CPU，内存，网络IO）。一般都是网络IO达到瓶颈。
1.5.3 Kafka的机器数量
Kafka机器数量=2*（峰值生产速度*副本数/100）+1
1.5.4 Kafka的日志保存时间：3天
1.5.5 Kafka的硬盘大小：每天的数据量*3天/70%
1.5.6 Kafka监控
公司自己开发的监控器；
开源的监控器：KafkaManager、KafkaMonitor、KafkaEagle
1.5.7 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个
1.5.8 副本数设定
一般我们设置成2个或3个，很多企业设置为2个。
1.5.9 多少个Topic
通常情况：多少个日志类型就多少个Topic。也有对日志类型进行合并的。
1.5.10 Kafka丢不丢数据
Ack=0，相当于异步发送，消息发送完毕即offset增加，继续生产。
Ack=1，leader收到leader replica 对一个消息的接受ack才增加offset，然后继续生产。
Ack=-1，leader收到所有replica 对一个消息的接受ack才增加offset，然后继续生产。
1.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中。
1.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进行排序，最后按照轮询的方式发给每一个消费线程。
1.5.13 Kafka中数据量计算
每天总数据量100g，每天产生1亿条日志， 10000万/24/60/60=1150条/每秒钟
平均每秒钟：1150条
低谷每秒钟：50条
高峰每秒钟：1150条*（2-20倍）=2300条-23000条
每条日志大小：0.5k-2k（取1k）
每秒多少数据量：2.0M-20MB
1.5.14 Kafka挂掉
1）Flume记录； 2）日志有记录； 3）短期没事
1.5.15 Kafka消息数据积压，Kafka消费能力不足怎么处理？
1）若Kafka消费能力不足，考虑增加Topic分区数，同时提升消费组的消费者数量，消费者数=分区数。（两者缺一不可）
2）如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间<生产速度），使处理的数据小于生产的数据，也会造成数据积压。
1.5.16 Kafka幂等性
Producer的幂等性指：当发送同一条消息时，数据在Server端只会被持久化一次，数据不丟不重，但这幂等性是有条件的：
1）只能保证Producer在单个会话内不丟不重，如果Producer出现意外挂掉再重启是无法保证的（幂等性情况下，是无法获取之前的状态信息，因此是无法做到跨会话级别的不丢不重）。
2）幂等性不能跨多个Topic-Partition，只能保证单个Partition内的幂等性，当涉及多个 Topic-Partition时，这中间的状态并没有同步。
1.5.17 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生命周期不同，同一事务的消息可能会出现重启后被删除的情况。
1.5.18 Kafka数据重复
幂等性+ack-1+事务
Kafka数据重复，可以再下一级：SparkStreaming、redis或者hive中dwd层去重，去重的手段：分组、按照id开窗只取第一个值；
1.5.19 Kafka参数优化
1）Broker参数配置（server.properties）
1、网络和io操作线程配置优化
# broker处理消息的最大线程数（默认为3）
num.network.threads=cpu核数+1
# broker处理磁盘IO的线程数
num.io.threads=cpu核数*2
2、log数据文件刷盘策略
# 每当producer写入10000条消息时，刷数据到磁盘
log.flush.interval.messages=10000
# 每间隔1秒钟时间，刷数据到磁盘
log.flush.interval.ms=1000
3、日志保留策略配置
# 保留三天，也可以更短 （log.cleaner.delete.retention.ms）
log.retention.hours=72
4、Replica相关配置
offsets.topic.replication.factor:3
# 这个参数指新创建一个topic时，默认的Replica数量,Replica过少会影响数据的可用性，太多则会白白浪费存储资源，一般建议在2~3为宜。
2）Producer优化（producer.properties）
buffer.memory:33554432 (32m)
#在Producer端用来存放尚未发送出去的Message的缓冲区大小。缓冲区满了之后可以选择阻塞发送或抛出异常，由block.on.buffer.full的配置来决定。
compression.type:none
#默认发送不进行压缩，推荐配置一种适合的压缩算法，可以大幅度的减缓网络压力和Broker的存储压力。
4）Kafka内存调整（kafka-server-start.sh）
默认内存1个G，生产环境尽量不要超过6个G。
export KAFKA_HEAP_OPTS="-Xms4g -Xmx4g"
1.5.20 Kafka高效读写数据
1）Kafka本身是分布式集群，同时采用分区技术，并发度高。
2）顺序写磁盘
Kafka的producer生产数据，要写入到log文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到600M/s，而随机写只有100K/s。这与磁盘的机械机构有关，顺序写快，是因为其省去了大量磁头寻址的时间。
3）零复制技术
1.5.21 Kafka支持传输
kafka对于消息体的大小默认为单条最大值是1M但是在我们应用场景中, 常常会出现一条消息大于1M，如果不对kafka进行配置。则会出现生产者无法将消息推送到kafka或消费者无法去消费kafka里面的数据, 这时我们就要对kafka进行以下配置：server.properties
1.5.22 Kafka过期数据清理
保证数据没有被引用（没人消费他）
日志清理保存的策略只有delete和compact两种
log.cleanup.policy=delete启用删除策略
log.cleanup.policy=compact启用压缩策略
1.5.23 Kafka可以按照时间消费数据
Map<TopicPartition, OffsetAndTimestamp> startOffsetMap = KafkaUtil.fetchOffsetsWithTimestamp(topic, sTime, kafkaProp);
1.5.24 Zookeeper中存储kafka哪些信息