Kafka(三)【Broker 存储】

目录

前言

Kafka Broker

1、工作流程

1.1、Zookeeper 存储的 Kafka 信息

1.2、Kafka Broker 的总体工作流程

1.3、Broke 重要参数

2、Kafka 副本

2.1、副本基本信息

2.2、Keader 选举流程

2.3、Leader 和 Follower 的故障处理细节

Follower 故障

Leader 故障:

2.4、分区副本分配

2.5、生产经验—手动调整分区副本存储

2.6、生产经验—Leader Partition 负载均衡

2.7、生产经验—增加副本因子

3、文件存储

3.1、文件存储机制

1)Topic 数据的存储机制

2)index 文件和log文件

3.2、文件清理策略

1)delete日志删除:将过期数据删除

2)compact 日志压缩

使用场景

4、高效读写数据(面试重点)

总结


前言

        今天学习 Kafka 的第二部分 Broker,相比较 Flink ,Kafka 的流式数据处理还是比较好理解的。

Kafka Broker

环境准备:

  • 启动 Zookeeper
  • 启动 Kafka

1、工作流程

我们先看看 Zookeeper 中存储了 Kafka 的哪些信息,它俩是如何协调工作的,

1.1、Zookeeper 存储的 Kafka 信息

我们需要在任意节点启动 Zookeeper 客户端:

# 启动zookeeper客户端
bin/zkCli.sh
# 查看信息
ls /kafka/brokers/ids

/kafka/brokers/ids 

可以看到,我们当前启动了三个 Kafka broker节点。

/kafka/brokers/topics/

这个目录下我们可以看到指定主题的分区信息:

/kafka/consumers/

  • 0.9 版本之前用于保存 offset 信息
  • 0.9版本之后 offset 存储在 kfaka 主题中

/kafka/controller

辅助节点选举,这个 controller 是谁取决于哪个节点先在 /kafka/controller 中注册的。在Kafka中,controller负责管理整个集群中所有分区和副本的状态。当检测到某个分区的ISR集合(ISR 就是所有 leader 和 follower 节点之间同步正常的节点集合)发生变化时,controller会负责通知所有broker更新其元数据信息。

1.2、Kafka Broker 的总体工作流程

  1. broker 节点启动后会先去 zookeeper 注册(注册到 /kafka/brokers/ids)代表自己活着
  2. 然后每个 broker 节点的 controller (每个 broker 节点都有一个 controller)会去zookeeper的 controller 注册一个临时节点(注册到 /kafka/controller),先注册的节点就是 controller ,因为只有一个Kafka Broker会注册成功,其他的都会失败,所以这个成功在Zookeeper上注册临时节点的Kafka Broker会成为Kafka Broker Controller,其他的Kafka broker 叫 Kafka Broker follower。
  3. 选举出来的 controller 会负责管理整个集群中所有分区和副本的状态
  4. controller 同时负责 Leader 的选举,选举规则是:首先要成为leader必须是存在于 ISR 中节点,其次按照这些节点在 AR 中的排名(AR 是 Kafak 分区中所有副本的统称),排在前面的节点优先成为 leader
  5. controller 节点将分区状态信息(当前集群的leade、isr ...)上传到 zookeeper (比如上传到 /kafka/brokers/topics/like/partitions/0/state)
  6. 其他 controller  去 zookeeper 拉取集群分区状态信息(防止 controller 挂了)
  7. 假设 leader 挂了
  8. controller 会立即检测到这个变化
  9. controller 从 zookeeper 中拉取回来集群分区状态信息,继续进行选举
  10. 选举的前提依然是必须存活于 ISR 队列,之后按照 AR 中的排列顺序,排在前面的优先成为 leader
  11. 选举完之后,controller 再次更新 zookeeper 中的集分区状态信息

1.3、Broke 重要参数

参数名称

描述

replica.lag.time.max.ms

ISR中,如果Follower长时间未向Leader发送通信请求或同步数据,则该Follower将被踢出ISR。该时间阈值,默认30s

auto.leader.rebalance.enable

默认是true。 自动Leader Partition 平衡。

leader.imbalance.per.broker.percentage

默认是10%。每个broker允许的不平衡的leader的比率。如果每个broker超过了这个值,控制器会触发leader的平衡。

leader.imbalance.check.interval.seconds

默认值300秒。检查leader负载是否平衡的间隔时间。

log.segment.bytes

Kafka中log日志是分成一块块存储的,此配置是指log日志划分 成块的大小,默认值1G。

log.index.interval.bytes

默认4kbkafka里面每当写入了4kb大小的日志(.log),然后就往index文件里面记录一个索引。

log.retention.hours

Kafka中数据保存的时间,默认7天。

log.retention.minutes

Kafka中数据保存的时间,分钟级别,默认关闭。

log.retention.ms

Kafka中数据保存的时间,毫秒级别,默认关闭。

log.retention.check.interval.ms

检查数据是否保存超时的间隔,默认是5分钟

log.retention.bytes

默认等于-1,表示无穷大。超过设置的所有日志总大小,删除最早的segment。

log.cleanup.policy

默认是delete,表示所有数据启用删除策略;

如果设置值为compact,表示所有数据启用压缩策略。

num.io.threads

默认是8。负责写磁盘的线程数。整个参数值要占总核数的50%

num.replica.fetchers

副本拉取线程数,这个参数占总核数的50%的1/3

num.network.threads

默认是3。数据传输线程数,这个参数占总核数的50%的2/3

log.flush.interval.messages

强制页缓存刷写到磁盘的条数,默认是long的最大值,9223372036854775807。一般不建议修改,交给系统自己管理。

log.flush.interval.ms

每隔多久,刷数据到磁盘,默认是null。一般不建议修改,交给系统自己管理。

2、Kafka 副本

2.1、副本基本信息

  1. Kafka副本作用:提高数据可靠性。
  2. Kafka默认副本1个,生产环境一般配置为2个,保证数据可靠性;太多副本会增加磁盘存储空间,增加网络上数据传输,降低效率。
  3. Kafka中副本分为:Leader和Follower。Kafka生产者只会把数据发往Leader,然后Follower找Leader进行同步数据。Kafka 中,生产者和消费者只能针对 Leader 进行操作
  4. Kafka分区中的所有副本统称为AR(Assigned Repllicas)。

 AR = ISR + OSR

ISR,表示和Leader保持同步的Follower集合。如果 Follower 长时间未向Leader发送通信请求或同步数据,则该Follower将被踢出ISR(踢出到 OSR)。该时间阈值由replica.lag.time.max.ms参数设定,默认30s。Leader发生故障之后,就会从 ISR 中选举新的Leader。

OSR表示Follower与Leader副本同步时,延迟过多的副本(特定条件下会恢复到 ISR)。

2.2、Keader 选举流程

        Kafka集群中有一个 broker 的 Controller 会被选举为 Controller Leader,负责管理集群broker的上下线,所有 topic 的分区副本分配Leader选举等工作。

        Controller的信息同步工作是依赖于Zookeeper的。

接下来我们来验证一下:

1. 创建一个新的topic,3个分区,3个副本

./kafka-topics.sh --bootstrap-server hadoop102:9092 -create --topic buy --partitions 3 -replication-factor 3

2. 查看 Leader 分布情况

注意区分 AR 和 ISR: 

 

3. 停止掉hadoop103 的 kafka 进程,并查看Leader分区情况

上面的图中我们可以看到hadoop103上的副本一共有 3 个 ,分别存在节点 [2,0,1] 上,并且当前的副本 leader 是 2,现在我们把 hadoop103 节点的 kafka 进程停掉,相当于保存在分区 1 的副本全部故障,而我们分区 3 中的副本的 leader 刚好是节点1,所以要把副本的 leader 进行更新,要按照理论应该是 2 号节点继续上任副本 leader 的位置。

 

我们发现,我们分区 3 的副本 leader  原本存在分区2 中,但是现在分区 2 挂掉了,所以它会从自己的 AR 中选一个在前面的节点 [2,0],于是 分区2 中的 副本 leader 变成了节点 2。

4. 我们再恢复一下hadoop103

[lyh@hadoop103 bin] ./kafka-server-start.sh -daemon ../config/server.properties 

2.3、Leader 和 Follower 的故障处理细节

概念:

  • LEO(Log End Offset):每个副本中最后一个 offset ,LEO 就是最新的 offset +1
  • HW(High Watermark):所有副本中最小的 LEO
Follower 故障

Leader 故障:

注意:只能保证副本之间的数据一致性,并不能保证数据不丢失或者不重复!

比如上面,我们的 leader 一开始是节点0,但是 leader 挂掉之后,leader 变成了节点1,但是新的leader 显然比旧的 leader 的数据少了3条,所以这部分数据就丢失了(毕竟自己处理问题总不能让人家生产者再发送一遍)。

2.4、分区副本分配

        如果kafka服务器只有3个节点,那么设置kafka 的分区数大于服务器台数,在kafka底层如何分配存储副本呢?

        注意:分区数可以大于机器数,毕竟大不了一个机器上多存几个分区,但是副本数不能大于机器数,那样没有意义,而且 Kafka 会报错:

我有三台机器,上面试着创建了 5 个副本的一个 topic,报错不能超过机器数量。

1)创建12分区,3个副本

./kafka-topics.sh -bootstrap-server hadoop102:9092 --create --topic card --partitions 12 --replication-factor 3

2)查看分区和副本信息

我们可以看到,Kafka 使得我们每个副本的 Leader 都尽可能的不一样,这样很好地分担了读写压力,毕竟Kafka 生产者和消费者都是只对分区 leader 进行操作的。每4个分区对应一个副本,而且每一个副本的 AR 顺序 Kafka 都尽量使它们不一样,这样可以尽可能做到负载均衡。

2.5、生产经验—手动调整分区副本存储

        生产环境中,每台服务器的配置和性能都不一致,但是 Kafka 只会根据自己的代码创建对应的分区副本,就会导致个别服务器存储压力比较大。所以需要手动调整分区副本的存储。

        需求:创建一个新的 topic ,4个分区,两个副本。我们将该 topic 的所有副本都存储到 hadoop102 和 hadoop103 上,hadoop104 不存储任何数据。

1)创建一个新的 topic 叫做 card ,4个分区,2个副本

2)查看分区副本信息

3)创建副本存储计划(所有副本都指定存储在hadoop102、hadoop103中)。

4)执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-repliaction-factor.json --execute

5)验证副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-repliaction-factor.json --verify

6)查看分区副本存储情况

2.6、生产经验—Leader Partition 负载均衡

        正常情况下,Kafka 本身会自动把 Leader Partition 均匀分散在各个机器上,来保证每台节点的吞吐量都是均匀的。但是如果某些 broker 节点宕机,会导致 leader partition 过于集中在其他少部分几台 broker 节点上,这会导致少数几台 broker 节点的读写压力过高,而且即使这些宕机的 broker 再次恢复上线,也只是 follower partition ,不会再次恢复为 leader partition,造成集群负载不均衡。

  • auto.leader.rebalance.enable,more为 true。自动 Leader partition 平衡
  • leader.imbalance.per.broker.percentag,默认是 10%,每个 broker 允许不平衡的leader的比率。如果超过这个值,控制器会触发 leader 的平衡。
  • leader.imbalance.check.interval.seconds,默认300s,也就是每300s检查一次leader负载是否平衡。

在生产环境中,通常不建议开启自动平衡,因为这可能会影响性能。

2.7、生产经验—增加副本因子

        假设我们创建了一个主题并设置副本数为 1 ,但是后来我们发现这部分数据特别重要,于是想要增加副本数,怎么办呢?通过命令行是不行的,得像我们上面 2.5 手动调整分区副本 一样,通过 json 文件来修改。

手动增加副本存储

1)创建副本存储计划(所有副本都指定存储在hadoop102、hadoop103、hadoop104中)

vim increase-replication-factor.json
# 输入下面的内容
{"version":1,"partitions":[{"topic":"four","partition":0,"replicas":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"topic":"four","partition":2,"replicas":[0,1,2]}]}

2)执行副本存储计划

bin/kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file increase-replication-factor.json --execute

3、文件存储

3.1、文件存储机制

1)Topic 数据的存储机制

        Topic是逻辑上的概念,而 partition 是物理上的概念,每个 partition 对应一个 log 文件,该 log 文件中存储的就是 producer 产生的数据。producer 产生的数据会被不断追加(追加是 Kafka 能够高效读写的一个重要原因)到该 log 文件末端,且每条数据都有自己的 offset 。消费者组中的每个消费者,都会实时记录自己消费到了哪个offset,以便出错恢复时,从上次的位置继续消费。为防止 log 文件过大导致数据定位效率低下,Kafka 采取了分区和索引机制,将每个 partition 分为多个 segment 。每个 segment 包括多个:“.index”文件、“.log”文件和 .timeindex 等文件(.timeindex 是一个时间戳索引文件,描述了文件保存时间,因为 Kafka 中的数据默认为保存 7 天才会自动删除,而判断文件日期是否达到7天就需要判断这个文件)。这些文件位于一个文件夹下,文件夹的命名规则为:topic名称+分区序号,例如:like-0。

# 000000~170409
00000000000000000000.index
00000000000000000000.log
# 170410~239429
00000000000000170410.index
00000000000000170410.log
# 239430~
00000000000000239430.index
00000000000000239430.log

验证:

1. 查看 kafka 的 data 目录下的 topic 数据

2. 这些文件因为都是经过序列化的所以都是乱码,需要使用 Kafka 提供的一个工具来看: 

# 查看 .index 文件
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000015.index 
# 查看 .log 文件
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000015.log

2)index 文件和log文件
  1. kafka 的 index 是稀疏索引,大约每往 .log 文件中写入 4KB 数据,会往 .index 文件写入一条索引。参数 log.index.interval.bytes 默认 4kb;
  2. index 文件中保存的 offset 是相对 offset,这样能确保 offset 的值所占空间不会太大,因此能将 offset 的值控制在固定大小。
  3. 一个 segment 文件大小 1GB

现有索引文件 000000.index,0000522.index,000001005.index ,如何在 log 文件中定位到 offset=600 的record?

  • offset=600,offset 大于522 小于 1005,说明就要找的文件索引就在 522.index 文件中,所以去 00000522.index 文件中
  • 000001005.index 文件中的数据有多行,因为每4kb数据就会往这个文件写一条记录

可以看到 587 就是我们的 index 文件名中的基础offset+相对offset得到的,计算得到 587 后我们发现要找的 offset=600 是大于 587 小于 639 的,说明要找的 record 就在这一行。于是查看 position 得到 6410,接着查看 log文件。

找到 000000522.index 对应的 000000522.log 文件,这个 log 文件同样有一个 position 属性,我们要找的 6410 刚好在log文件中就有一条记录的position= 6410,这就找到了。如果我们要找的position=6415,那么我们就得找到介于这个值中间的数据,因为 6410 < 6415 <10090 所以我们要找的 position=6415的数据就在 6410这一行。

3.2、文件清理策略

Kafka中默认的日志保存时间为7,可以通过调整如下参数修改保存时间。

  1. log.retention.hours(int),最低优先级小时,默认7天。
  2. log.retention.minutes(int),分钟。
  3. log.retention.ms(long),最高优先级毫秒。
  4. log.retention.check.interval.ms,负责设置检查周期,默认5分钟。

这四个参数的优先级从上到下越来越高,也就是说 当我们设置了日志保存参数为 ms 级别时,前面设置的 hours 和 minutes 级别的参数就都失效了。

那么日志一旦超过了设置的时间,怎么处理呢?

Kafka中提供的日志清理策略有 delete 和 compact 两种。

1)delete日志删除:将过期数据删除
  • log.cleanup.policy = delete    所有数据启用删除策略

(1)基于时间:默认打开以segment中所有记录中的最大时间戳作为该文件时间戳。

(2)基于大小:默认关闭。超过设置的所有日志总大小,删除最早的segment。

log.retention.bytes(long),默认等于-1,表示无穷大。

思考:如果一个segment中有一部分数据过期,一部分没有过期,怎么处理?

 当然是以segment中所有记录中的最大时间戳作为该文件时间戳,所以即使数据有99.9%是旧的,只要有0.01%是新的数据,就得等它过期了才能删除。

2compact 日志压缩

compact 日志压缩:对于相同的 key 的不同 value 只保留最后一个版本

要使用这个功能,只需要修改配置

  • log.cleanup.policy=compact

使用场景

        这种压缩只能用于特定场景,比如消息的 key 是用户id,value是用户的资料,通过这个压缩,整个消息集里就保存了所有用户最新的信息。

4、高效读写数据(面试重点)

1Kafka本身是分布式集群,可以采用分区技术,并行度高

2)读数据采用稀疏索引,可以快速定位要消费的数据

3)顺序写磁盘

        Kafka的producer生产数据,要写入到log文件中,写的过程是一直追加到文件末端,为顺序写。官网有数据表明,同样的磁盘,顺序写能到600M/s,而随机写只有100K/s。这与磁盘的机械结构有关,顺序写之所以快,是因为其省去了大量磁头寻址的时间。

4)页缓存 + 零拷贝技术

零拷贝:Kakfa 把对数据操作的步骤放到了 生产者和消费者当中(生产者和消费者可以在拦截器来对数据进行处理),所以 Kafka Broker 应用层并不关心数据的存储,所以数据都不需要走应用层,直接走网卡就可以传输费 消费者,传输效率高

pageCache 页缓存:Kafka 重度依赖底层操作系统提供的 PageCache 功能。当上层有写操作时,操作系统只是将数据写入 pageCache。当读操作发生时,先从 pageCache 中查找,如果找不到,再去磁盘读取。实际上 pageCache 是尽可能把更多的空闲内存空间都当做了磁盘缓存来使用。

参数

描述

log.flush.interval.messages

强制页缓存刷写到磁盘的条数,默认是long的最大值,9223372036854775807。一般不建议修改,交给系统自己管理。

log.flush.interval.ms

每隔多久,刷数据到磁盘,默认是null。一般不建议修改,交给系统自己管理。

总结

        这一节用到的 prettyZoo 挺震撼我的,JavaFX 能开发出如此漂亮使用的一款软件 ,实在让我想不到,希望自己写的软件有一天也可以为百千人使用。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/422413.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

每日一题——LeetCode1299.将每个元素替换为右侧最大元素

方法一 个人方法&#xff1a; 题目意思就是求在i1;i的循环条件下&#xff0c;arr[i]-arr[arr.length-1]的最大值分别为多少&#xff0c;最后一项默认为-1 用slice方法可以每次把数组第一位去除&#xff0c;得到求最大值的目标数组 Math的max方法可以直接返回数组里的最大值 …

<C++>STL->list

list的介绍 list list是一个类模板&#xff0c;第一个模板参数为存储数据类型&#xff1b;第二个模板参数为空间适配器list是一个可以在常数时间内完成任意位置的插入和删除的顺序容器。list容器是以双链表的形式实现的&#xff1b;双链表可以将其包含的每个元素存储在不同且…

批量转换:速度与质量的完美结合

你是否曾经遇到过需要批量转换上千条视频格式的挑战&#xff1f;这种任务不仅耗时&#xff0c;而且需要专业的视频处理技术。那么&#xff0c;有没有一种简单而高效的方法来完成这个任务呢&#xff1f;答案是肯定的。接下来&#xff0c;我将为你介绍一种快速、稳定的一分钟批量…

VMware workstation平台下配置Fedora-Server-39-1.5虚拟机网络

VMware workstation平台下配置Fedora-Server-39-1.5虚拟机网络 Fedora包含的软件以自由及开放源码许可来发布&#xff0c;并旨在成为该技术领域的领先者。Fedora在专注创新、抢先集成新技术、与上游Linux社区紧密工作方面拥有良好名声。该文档适用于在VMware workstation平台下…

meshgrid contour contourf

meshgrid contour contourf 参考video: https://www.bilibili.com/video/BV1qW411A775/?spm_id_from333.337.search-card.all.click&vd_sourced171c31a396363b8ea8c0e92a59cee6b 官方文档: https://matplotlib.org/stable/api/_as_gen/matplotlib.pyplot.contourf.html#ma…

网络要素服务(WFS)详解

文章目录 1. 概述2. GetCapabilities3. DescribeFeatureType4. GetFeature4.1 Get访问方式4.2 Post访问方式 5. Transaction5.1 Insert5.2 Replace5.3 Update5.4 Delete 6 注意事项 1. 概述 前置文章&#xff1a; 地图服务器GeoServer的安装与配置 GeoServer发布地图服务&#…

Postman如何做接口测试:如何导入 swagger 接口文档

&#x1f525; 交流讨论&#xff1a;欢迎加入我们一起学习&#xff01; &#x1f525; 资源分享&#xff1a;耗时200小时精选的「软件测试」资料包 &#x1f525; 教程推荐&#xff1a;火遍全网的《软件测试》教程 &#x1f4e2;欢迎点赞 &#x1f44d; 收藏 ⭐留言 &#x1…

透明拼接屏在汽车领域的应用

随着科技的进步&#xff0c;透明拼接屏作为一种新型的显示技术&#xff0c;在汽车领域的应用越来越广泛。尼伽小编将围绕透明拼接屏在汽车本身、4S店、展会、工厂等方面的应用进行深入探讨&#xff0c;并展望未来的设计方向。 一、透明拼接屏在汽车本身的应用 车窗显示&#x…

Unity工程没有创建.sln文件,导致打开C#文件无法打开解决方案

最近又开始折腾些Unity的小项目&#xff0c;重新遇到一些常见的小问题 点击报错文件 却没有打开文件 于是查看了下打开Window->Package Manager 选择Unity Registry 搜索Visual Studio Editor&#xff0c;发现并没有安装 同理&#xff0c;也可以安装VSCode的插件 问题解决了…

【BBuf的CUDA笔记】十三,OpenAI Triton 入门笔记一

0x0. 前言 2023年很多mlsys工作都是基于Triton来完成或者提供了Triton实现版本&#xff0c;比如现在令人熟知的FlashAttention&#xff0c;大模型推理框架lightllm&#xff0c;diffusion第三方加速库stable-fast等灯&#xff0c;以及很多mlsys的paper也开始使用Triton来实现比…

谁适合选择虚拟化

情况 前些天,有人问弄虚拟化怎么样: 还有一个群里,讨论了这事: 也弄了很多年了,虽然不算深入,毕竟,也是拼尽了全力,毕竟差不多7年的时光已经投入进去了,回头时,感觉没留下什么,有些十年技术一场空的感觉,真是应了虚拟化这几个字。 现在就大体说说这事: 先看看当前…

基于TableAgent实现IT职位招聘数据分析—以传统机器学习与TableAgent 数据分析方式相对比以凸显TableAgent 特性

目录 &#x1f680;一. TableAgent—新AI时代的数据分析智能体 &#x1f50e;1.1 基于DataCanvas Alaya九章元识大模型 &#x1f50e;1.2 TableAgent的亮点 &#x1f680;二. 使用TableAgent分析数据与传统机器学习分析数据对比 &#x1f50e;2.1 项目背景 &#x1f50e;2.2 数…