四十、大数据技术之Kafka3.x（3）

devtools/2024/11/14 22:46:12/

🌻🌻 目录

  • 一、Kafka Broker
    • 1.1 Kafka Broker工作流程
      • 1.1.1 Zookeeper 存储的Kafka信息
      • 1.1.2 Kafka Broker 总体工作流程
      • 1.1.3 Broker 重要参数
    • 1.2 生产经验——节点服役和退役
      • 1.2.1 服役新节点
      • 1.2.2 退役旧节点
    • 1.3 Kafka 副本
      • 1.3.1 副本基本信息
      • 1.3.2 Leader 选举流程
      • 1.3.3 Leader 和 Follower 故障处理细节
      • 1.3.4 分区副本分配
      • 1.3.5 生产经验——手动调整分区副本存储
      • 1.3.6 生产经验——Leader Partition负载平衡
      • 1.3.7 生产经验——增加副本因子
    • 1.4 文件存储
      • 1.4.1 文件存储机制
      • 1.4.2 文件清理策略
    • 1.5 高效读写数据

一、Kafka Broker

1.1 Kafka Broker工作流程

1.1.1 Zookeeper 存储的Kafka信息

（1）启动Zookeeper客户端。

cd /usr/local/zookeeper/bin/ls./zkCli.sh

在这里插入图片描述

（2）通过ls命令可以查看kafka相关信息。

ls /ls /kafka/ls /kafka/brokers/ls /kafka/brokers/ids

在这里插入图片描述

zookeerper 可视化工具 PrettyZoo：

  • 下载
  • 使用参考

①

在这里插入图片描述

②

在这里插入图片描述

在这里插入图片描述

1.1.2 Kafka Broker 总体工作流程

在这里插入图片描述

1）模拟Kafka上下线，Zookeeper中数据变化

（1）查看/kafka/brokers/ids路径上的节点。

./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties./kafka-server-stop.sh

在这里插入图片描述

在这里插入图片描述

1.1.3 Broker 重要参数

在这里插入图片描述

1.2 生产经验——节点服役和退役

1.2.1 服役新节点

1）新节点准备
（1）关闭linux-102(已经装了hadoop,jdk,zookeeper,kafka)，并右键执行克隆操作,linux-103,linux-104(注：之前若有直接可以删掉重新克隆)
（2）开启linux-103，linux-104 并修改IP地址。

①

在这里插入图片描述

②

在这里插入图片描述

③

在这里插入图片描述

④ 修改ip，将192.168.10.102修改为 192.168.10.103，保存，并重启网络

在这里插入图片描述

在这里插入图片描述

⑤ 修改 主机名

在这里插入图片描述
在这里插入图片描述

⑥ 重启进行远程连接

在这里插入图片描述

⑦ 删除日志，并分别修改linux-103与linux-102 的节点 id 和服务器 ip，并开启集群配置

在这里插入图片描述

在这里插入图片描述

开启集群配置(linux-102,linux-103,linux-104都需修改)

在这里插入图片描述

(3) 先启动linux-102,再启动linux-103，再次启动 linux-104(脚本启动后期更新)

cd /usr/local/zookeeper/bin/./zkServer.sh start./zkServer.sh statuscd /usr/local/kafka/bin./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

(4) 查看 linux-102 上有哪些主题，并分配了分区(生产环境建议至少分两个副区，如果不小心删除一个，则还有一个类似备份的)

./kafka-topics.sh --bootstrap-server linux-102:9092 --list./kafka-topics.sh --bootstrap-server linux-102:9092 --topic first --describe

在这里插入图片描述

思考：如何将linux-102上面的一些分到linux-103达到负载均衡呢？看下面 2）

2）执行负载均衡操作

（1）创建一个要均衡的主题（在 linux-102 服务器的 kafka下面创建）。

cd /usr/local/kafkavi topics-to-move.json

在这里插入图片描述

{"topics": [{"topic": "first"}],"version": 1
}

在这里插入图片描述

（2）生成一个负载均衡的计划（在 linux-102 服务器的 kafka下面生成）。

# 0,1,2 代表三台服务器
bin/kafka-reassign-partitions.sh --bootstrap-server linux-102:9092  --topics-to-move-json-file topics-to-move.json --broker-list "0,1,2" --generate

遇到错误：java.io.IOException: Unable to read file topics-to-move.json

在这里插入图片描述

解决：直接切换到kafka 根目录进行生成即可：

在这里插入图片描述

（3）创建副本存储计划（所有副本存储在broker2、broker3、broker4）。

vi increase-replication-factor.json

在这里插入图片描述

输入如下内容：

{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":1,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":3,"replicas":[2],"log_dirs":["any"]}]}

在这里插入图片描述

（4）执行副本存储计划。

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

在这里插入图片描述

（5）验证副本存储计划。

bin/kafka-reassign-partitions.sh --bootstrap-server linux-102:9092 --reassignment-json-file increase-replication-factor.json --verify

在这里插入图片描述

上述操作都是在linux-102上操作的。

1.2.2 退役旧节点

1）执行负载均衡操作

先按照退役一台节点，生成执行计划，然后按照服役时操作流程执行负载均衡。

（1）创建一个要均衡的主题。

与上面（1）创建一个要均衡的主题（在 linux-102 服务器的 kafka下面创建）。一样无需再创建。

（2）创建执行计划。

#上面是分配到了三台机器 2，3，4，现在移除 4即服务器 linux-104
bin/kafka-reassign-partitions.sh --bootstrap-server linux-102:9092 --topics-to-move-json-file topics-to-move.json --broker-list "2,3" --generate

在这里插入图片描述

（3）创建副本存储计划（所有副本存储在broker2、broker3）。

vi increase-replication-factor.json

在这里插入图片描述

删除前面的，将上面复制的粘贴里面保存即可。

在这里插入图片描述

粘贴上面复制的如下内容：

{"version":1,"partitions":[{"topic":"first","partition":0,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":1,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":2,"replicas":[2],"log_dirs":["any"]},{"topic":"first","partition":3,"replicas":[2],"log_dirs":["any"]}]}

（4）执行副本存储计划。

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

在这里插入图片描述

（5）验证副本存储计划。

bin/kafka-reassign-partitions.sh --bootstrap-server linux-102:9092 --reassignment-json-file increase-replication-factor.json --verify

在这里插入图片描述
2）执行停止命令

在linux-104上执行停止命令即可

./kafka-server-stop.sh

在这里插入图片描述

1.3 Kafka 副本

下面的所有后期会详细更新（可跳过直接看 👉👉大数据技术之Kafka3.x（4））

1.3.1 副本基本信息

  • （1）Kafka 副本作用：提高数据可靠性。
  • （2）Kafka默认副本1个，生产环境一般配置为2个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
  • （3）Kafka中副本分为：Leader和Follower。Kafka生产者只会把数据发往Leader，然后Follower找Leader进行同步数据。
    （4）Kafka分区中的所有副本统称为AR（Assigned Repllicas）。
    AR = ISR + OSR
    ISR，表示和Leader保持同步的Follower集合。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。Leader发生故障之后，就会从ISR中选举新的Leader。
    OSR，表示Follower与Leader副本同步时，延迟过多的副本。

1.3.2 Leader 选举流程

(跳过后期更新)

Kafka 集群中有一个broker的Controller会被选举为Controller Leader，负责管理集群 broker 的上下线，所有topic的分区副本分配和Leader选举等工作。
Controller的信息同步工作是依赖于Zookeeper的。

在这里插入图片描述

开启 linux-102,linux-103,linux-104,进行如下操作：

（1）创建一个新的topic，4个分区，4个副本（因为我是单节点，所以设置了 1）

bin/kafka-topics.sh --bootstrap-server linux-102:9092 --create --topic Daniel1 --partitions 1 --replication-factor 1

在这里插入图片描述

（2）查看Leader分布情况

bin/kafka-topics.sh --bootstrap-server linux-102:9092 --describe --topic Daniel1

在这里插入图片描述

（3）停止掉linux-103的kafka进程，并查看Leader分区情况

在这里插入图片描述

（4）停止掉linux-104的kafka进程，并查看Leader分区情况
（5）启动linux-104的kafka进程，并查看Leader分区情况
（6）启动linux-104的kafka进程，并查看Leader分区情况
（7）停止掉linux-103的kafka进程，并查看Leader分区情况

1.3.3 Leader 和 Follower 故障处理细节

在这里插入图片描述

1.3.4 分区副本分配

(跳过后期更新)

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

1）创建16分区，3个副本

（1）创建一个新的topic，名称为second。

bin/kafka-topics.sh --bootstrap-server linux-102:9092 --create --topic second --partitions 16 --replication-factor 3

在这里插入图片描述

（2）查看分区和副本情况。

在这里插入图片描述

在这里插入图片描述

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

在这里插入图片描述

手动调整分区副本存储的步骤如下：

（1）创建一个新的topic，名称为three。

在这里插入图片描述

（2）查看分区副本存储情况。

在这里插入图片描述

（3）创建副本存储计划（所有副本都指定存储在broker0、broker1中）。
（4）执行副本存储计划。
（5）验证副本存储计划。
（6）查看分区副本存储情况。

1.3.6 生产经验——Leader Partition负载平衡

在这里插入图片描述

在这里插入图片描述

1.3.7 生产经验——增加副本因子

在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的增加需要先制定计划，然后根据计划执行。

1）创建topic
2）手动增加副本存储
（1）创建副本存储计划（所有副本都指定存储在broker0、broker1、broker2中）。
（2）执行副本存储计划。

1.4 文件存储

1.4.1 文件存储机制

1）Topic数据的存储机制

在这里插入图片描述

2）思考：Topic数据到底存储在什么位置？

（1）启动生产者，并发送消息。
（2）查看linux-102（或者linux-103、linux-104）的/usr/local/kafka/datas/first-1（first-0、first-2）路径上的文件。
（3）直接查看log日志，发现是乱码。
（4）通过工具查看index和log信息。
3）index文件和log文件详解

在这里插入图片描述

说明：日志存储参数配置

在这里插入图片描述

1.4.2 文件清理策略

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

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

  • log.retention.hours，最低优先级小时，默认7天。
  • log.retention.minutes，分钟。
  • log.retention.ms，最高优先级毫秒。
  • log.retention.check.interval.ms，负责设置检查周期，默认5分钟。

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

Kafka中提供的日志清理策略有delete和compact两种。
1）delete日志删除：将过期数据删除

  • log.cleanup.policy = delete 所有数据启用删除策略
    （1）基于时间：默认打开。以segment中所有记录中的最大时间戳作为该文件时间戳。
    （2）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的segment。
    log.retention.bytes，默认等于-1，表示无穷大。

思考：如果一个segment中有一部分数据过期，一部分没有过期，怎么处理？

在这里插入图片描述

2）compact日志压缩

在这里插入图片描述

1.5 高效读写数据

  • 1）Kafka本身是分布式集群，可以采用分区技术，并行度高
  • 2）读数据采用稀疏索引，可以快速定位要消费的数据
  • 3）顺序写磁盘

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

在这里插入图片描述

4）页缓存 + 零拷贝技术

在这里插入图片描述


http://www.ppmy.cn/devtools/94179.html

相关文章

vue RSA加密解密(解决加密过长，解密过长返回为null的问题)

vue RSA加密解密(解决加密过长，解密过长返回为null的问题)

1安装 npm i jsencrypt2.rsa.js /* 引入jsencrypt实现数据RSA加密 */ import JSEncrypt from jsencrypt // 处理长文本数据时报错 jsencrypt.js Message too long for RSA /* 引入encryptlong实现数据RSA加密 */ //import Encrypt from encryptlong // encryptlong是基于jsen…
阅读更多...
java实现七牛云内容审核功能，文本、图片和视频的内容审核（鉴黄、鉴暴恐、敏感人物）

java实现七牛云内容审核功能，文本、图片和视频的内容审核（鉴黄、鉴暴恐、敏感人物）

目录 1、七牛云内容审核介绍 2、查看内容审核官方文档 2.1、文本内容审核 2.1.1、文本内容审核的请求示例 2.1.2、文本内容审核的返回示例 2.2、图片内容审核 2.2.1、请求参数 2.2.2、返回参数 2.3、视频内容审核 3、代码实现 3.1、前期代码准备 3.2、文本内容审核…
阅读更多...
智能合约语言对比：Solidity | Vyper | Move | Rust

智能合约语言对比：Solidity | Vyper | Move | Rust

当你想要进入Web3领域做开发，可能会想知道应该先学哪门编程语言，或者是哪门语言最适合你。这里有四种现在比较热门的语言：Solidity、Vyper、Move和Rust。下面我会用代码示例解释一下区别，帮你找到学习的方向。 Solidity&#xff1…
阅读更多...
Node.js中判断是文件还是文件夹的多种方法

Node.js中判断是文件还是文件夹的多种方法

在Node.js中，我们经常需要判断一个路径是文件还是文件夹。Node.js提供了多种方法来实现这一功能，本文将详细介绍这些方法，并给出相应的示例代码。 一、使用fs.Stats对象 在Node.js中，fs模块提供了fs.stat()或fs.statSync()方法&…
阅读更多...
【日记】总感觉今天很感性呢（2424 字）

【日记】总感觉今天很感性呢（2424 字）

正文 高温橙色预警。我并没有受到影响，因为今天上班…… 今天又和柜面主管吵了起来。两个人观念上的不同。今天装了三本档案，她嫌弃我装得不好，说不好看。当时她反问我信贷档案也是这样装的吗？装得这么随便。当时我跟她说&#xf…
阅读更多...
停止项目大小调整，开始搜索层自动缩放！

停止项目大小调整，开始搜索层自动缩放！

作者：来自 Elastic Matteo Piergiovanni，John Verwolf 我们新的 serverless 产品的一个关键方面是允许用户部署和使用 Elastic，而无需管理底层项目节点。为了实现这一点，我们开发了搜索层自动扩展，这是一种根据我们将在…
阅读更多...
数字取证：解密信息安全的科技利剑

数字取证：解密信息安全的科技利剑

标题：数字取证：解密信息安全的科技利剑 在信息时代，数据成为了一种强大的证据形式，数字取证作为一门科学，专注于电子证据的收集、分析和报告。它不仅在法律领域有着举足轻重的作用，更在企业安全、网络入侵…
阅读更多...
JAVA中BigInteger的基础用法

JAVA中BigInteger的基础用法

BigInteger 概述 BigInteger 是 Java 中的一个类，位于 java.math 包下。它提供了任意精度的整数运算能力，即它可以处理那些超出 int 和 long 类型表示范围的整数。BigInteger 是不可变的，即一旦一个 BigInteger 对象被创建，它的值…
阅读更多...
最新文章

Copyright @ 2022~2023