Kafka(六)Kafka基本客户端命令操作

转载自:https://blog.51cto.com/littledevil/2147950

主题管理

创建主题

如果配置了auto.create.topics.enable=true(这也是默认值)这样当生产者向一个没有创建的主题发送消息就会自动创建,其分区数量和副本数量也是有默认配置来控制的。

# 我们这里创建一个3个分区每个分区有2个副本的主题
kafka-topics.sh --create --zookeeper 172.16.48.171:2181/kafka --replication-factor 2 --partitions 3 --topic KafkaTest
--create表示建立
--zookeeper

表示ZK地址,可以传递多个,用逗号分隔

--zookeeper IP:PORT,IP:PORT,IP:PORT/kafka

--replication-factor表示副本数量,这里的数量是包含Leader副本和Follower副本,副本数量不能超过代理数量
--partitions表示主题的分区数量,必须传递该参数。Kafka的生产者和消费者采用多线程并行对主题的消息进行处理,每个线程处理一个分区,分区越多吞吐量就会越大,但是分区越多也意味着需要打开更多的文件句柄数量,这样也会带来一些开销。
--topic表示主题名称

image.png

在Zookeeper中可以看到如下信息

image.png

删除主题

删除有两种方式手动和自动

  • 手动方式需要删除各个节点日志路径下的该主题所有分区,并且删除zookeeper上/brokers/topics和/config/topics下的对应主题节点

  • 自动删除就是通过脚本来完成,同时需要配置服务器配置文件中的delete.topic.enable=true,默认为false也就是说通过命令删除主题只会删除ZK中的节点,日志文件不会删除需要手动清理,如果配置为true,则会自动删除日志文件。

kafka-topics.sh --delete --zookeeper 172.16.48.171:2181/kafka --topic KafkaTest

image.png

下面的两句话就是说该主题标记为删除/admin/delete_topics节点下。实际数据没有影响因为该参数没有设置为true。

查看主题

# 列出所有主题
kafka-topics.sh --list --zookeeper 172.16.48.171:2181/kafka

image.png

下面是从ZK中看到的所有主题

image.png

# 查看所有主题信息
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka

image.png

# 查看特定主题信息
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --topic BBB

image.png

Replicas:是AR列表,表示副本分布在哪些代理上,且该列表第一个元素就是Leader副本所在代理

ISR:该列表是显示已经同步的副本集合,这个列表的副本都是存活的

# 通过--describe 和 --under-replicated-partitions 可以查看正在同步的主题或者同步可能发生异常,
# 也就是ISR列表长度小于AR列表,如果一切正常则不会返回任何东西,也可以通过 --tipic 指定具体主题
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --under-replicated-partitions
# 查看哪些主题建立时使用了单独的配置 
kafka-topics.sh --describe --zookeeper 172.16.48.171:2181/kafka --topics-with-overrides

这里只有一个内部主题__comsumer_offsets使用了非配置文件中的设置

image.png

 

配置管理

所谓配置就是参数,比如修改主题的默认参数。

主题级别的

# 查看配置
kafka-configs.sh --describe --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BB

这里显示 Configs for topic 'BBB' are 表示它的配置有哪些,这里没有表示没有为该主题单独设置配置,都是使用的默认配置。

image.png

# 增加一个配置
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BBB --alter --add-config flush.messages=2

image.png

如果修改的话还是相同的命令,只是把值修改一下

image.png

# 删除配置
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --entity-type topics --entity-name BBB --alter --delete-config flush.messages

image.png

客户端级别

这个主要是设置流控

# 设置指定消费者的流控 --entity-name 是客户端在创建生产者或者消费者时是指定的client.id名称
kafka-configs.sh --zookeeper 172.16.48.171:2181/kafka --alter --add-config 'producer_byte_rate=1024,consumer_byte_rate=1024' --entity-type clients --entity-name COMSUMER_NAME

image.png

下图为ZK中对应的信息

image.png

 

分区管理

分区平衡

Leader副本在集群中应该是均衡分布,因为Leader副本对外提供读写服务,尽可能不让同一个主题的多个Leader副本在同一个代理上,但是随着时间推移比如故障转移等情况发送,Leader副本可能不均衡。有两种方式设置自动平衡,自动和手动。

自动就是在配置文件中增加 auto.leader.rebalance.enable true 如果该项为false,当某个节点故障恢复并重新上线后,它原来的Leader副本也不会转移回来,只是一个Follower副本。

手动就是通过命令来执行

kafka-preferred-replica-election.sh --zookeeper 172.16.48.171:2181/kafka

分区迁移

当下线一个节点需要将该节点上的分区副本迁移到其他可用节点上,Kafka并不会自动进行分区迁移,如果不迁移就会导致某些主题数据丢失和不可用的情况。当增加新节点时,只有新创建的主题才会分配到新节点上,之前的主题分区不会自动分配到新节点上,因为老的分区在创建时AR列表中没有这个新节点。

image.png

上面2个主题,每个主题3个分区,每个分区3个副本,我们假设现在代理2要下线,所以我们要把代理2上的这两个主题的分区数据迁移出来。

# 1. 在KAFKA目录的config目录中建立topics-to-move.json文件
{"topics":[{"topic":"AAA"},{"topic":"BBB"}],"version":1
}

 

# 2. 生成分区分配方案,只是生成一个方案信息然后输出
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --topics-to-move-json-file ./topics-to-move.json --broker-list "1,2" --generate

image.png

这个命令的原理是从zookeeper中读取主题元数据信息及制定的有效代理,根据分区副本分配算法重新计算指定主题的分区副本分配方案。把【Proposed partition reassignment configuration】下面的分区方案保存到一个JSON文件中,partitions-reassignment.json 文件名无所谓。

# 3. 执行方案
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --execute

image.png

# 4. 查看进度
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --verify

image.png

查看结果,这里已经没有代理0了。

image.png

集群扩容

上面演示了节点下线的数据迁移,这里演示一下集群扩容的数据迁移。我们还是用上面两个主题,假设代理0又重新上线了。其实扩容就是上面的反向操作

# 1. 建立JSON文件
# 该文件和之前的相同

 

# 2. 生成方案并保存到一个JSON文件中
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --topics-to-move-json-file ./topics-to-move.json --broker-list "0,1,2" --generate

image.png

# 3. 数据迁移,这里通过--throttle做一个限流操作,如果数据过大会把网络堵塞。
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./partitions-reassignment.json --execute --throttle 1024

image.png

查看进度和结果

image.png

增加分区

通常在需要提供吞吐量的时候我们会增加分区,然后如果代理数量不扩大,同时生产者和消费者线程不增大,你扩展了分区也没有用。

image.png

kafka-topics.sh --alter --zookeeper 172.16.48.171:2181/kafka --partitions 3 --topic KafkaTest03

image.png

增加副本

集群规模扩大并且想对所有主题或者指定主题提高可用性,那么可以增加原有主题的副本数量

image.png

上面是3个分区,每个分区1个副本,我们现在把每个分区扩展为3个副本

# 1. 创建JSON文件 replica-extends.json 
{"version": 1,"partitions": [{"topic": "KafkaTest04","partition": 0,"replicas": [0,1,2]},{"topic": "KafkaTest04","partition": 1,"replicas": [0,1,2]},{"topic": "KafkaTest04","partition": 2,"replicas": [0,1,2]}]
}
# 2. 执行分区副本重新分配命令
kafka-reassign-partitions.sh --zookeeper 172.16.48.171:2181/kafka --reassignment-json-file ./replica-extends.json --execute

image.png

查看状态

image.png

查看结果

image.png

 

镜像操作

Kafka有一个镜像工具kafka-mirror-maker.sh,用于将一个集群数据同步到另外一个集群中,这个非常有用,比如机房搬迁就需要进行数据同步。该工具的本质就是创建一个消费者,在源集群中需要迁移的主题消费数据,然后创建一个生产者,将消费的数据写入到目标集群中。

首先创建消费者配置文件mirror-consumer.properties(文件路径和名称是自定义的)

# 源kafka集群代理地址列表
bootstrap.servers=IP1:9092,IP2:9092,IP3:9092
# 消费者组名
group.id=mirror

其次创建生产者配置文件mirror-producer.properties(文件路径和名称是自定义的)

# 目标kafka集群地址列表
bootstrap.servers=IP1:9092,IP2:9092,IP3:9092

运行镜像命令

# 通过 --whitelist 指定需要镜像的主题,通过  --blacklist 指定不需要镜像的主题
kafka-mirror-maker.sh --consumer.config PATH/mirror-consumer.properties --producer.config PATH/mirror-producer.properties --whitelist TOPIC

由于镜像操作是启动一个生产者和消费者,所以数据同步完成后这个生产者和消费者并不会关闭,它会依然等待新数据,所以同步完成以后你需要自己查看,确认完成了则关闭生产者和消费者。

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

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

相关文章

linux平台C++开发基本知识

最近工作中,需要在linux上开发C程序。有了下面的问题。 1,linux平台C开发和win32 C开发有什么区别呢? 2,除了C语言,数据结构等必须的知识外,还需要些了解什么呢? 3,如何在linux平台开…

Impala 调用Hbase 报错 LeaseException

impala调用Hbase表数据操作时,提示 LeaseException 异常,主要原因是hbase数据过大,调用期间没有汇报心跳导致 WARNINGS: LeaseException: org.apache.hadoop.hbase.regionserver.LeaseException: lease -8355984789923245890 does not exist…

js和jsp所有页面跳转总结

Jsp页面跳转和js控制页面跳转的几种方法 Jsp 页面跳转的几种方法 1. RequestDispatcher.forward() 在服务器端起作用,当使用forward()时,Servlet engine传递HTTP请求从当前的Servlet或者是JSP到另外的一个Servlet、JSP 或普通HTML文件,也即你的form提交至a.jsp,在a.jsp用到了fo…

mitmproxy 中间人代理工具,抓包工具,linux抓包工具 mitmproxy 使用

mitmproxy是一个支持HTTP和HTTPS的抓包程序,类似Fiddler、Charles的功能,可以在linux以命令行形式的展示抓包信息 mitmdump:它是mitmproxy的命令行接口,利用它我们可以对接Python脚本,用Python实现监听后的处理。 mitm…

欧拉角转旋转矩阵

//c实现 cv::Mat eulerAnglesToRotationMatrix(cv::Vec3f& theta){// Calculate rotation about x axiscv::Mat R_x (cv::Mat_<double>(3, 3) <<1, 0, 0,0, cos(theta[0]), -sin(theta[0]),0, sin(theta[0]), cos(theta[0]));// Calculate rotation about y a…

C++常见面试题

1.在C 程序中调用被C 编译器编译后的函数&#xff0c;为什么要加extern “C”&#xff1f; 答&#xff1a;首先&#xff0c;extern是C/C语言中表明函数和全局变量作用范围的关键字&#xff0c;该关键字告诉编译器&#xff0c;其声明的函数和变量可以在本模块或其它模块中使用。…

Ajax学习总结+案例

一、AJAX简介 1、Asynchronous JavaScript And XML指异步 JavaScript 及 XML 2、老技术新用法。是基于JavaScript、XML、HTML、CSS新用法 二、同步和异步&#xff08;理解&#xff09; 三、第一个异步请求案例&#xff08;熟悉编码步骤&#xff09; 四、XmlHttpRequest&…

spark 2.2 读取 Hadoop3.0 数据异常 org.apache.hadoop.hdfs.web.HftpFileSystem cannot access its superinterfa

spark 2.2 读取 Hadoop3.0 数据异常 Exception in thread "main" java.lang.IllegalAccessError: class org.apache.hadoop.hdfs.web.HftpFileSystem cannot access its superinterface org.apache.hadoop.hdfs.web.TokenAspect$TokenManagementDelegatorat java.la…

安装open3d python

安装&#xff1a; pip install open3d 测试是否安装成功 python -c "import open3d as o3d"

ajax测试代码

//jsp文件 <% page language"java" contentType"text/html; charsetutf-8"pageEncoding"utf-8"%> <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd"> …

Spark 运行内存不足Not enough space to cache rdd in memory,Container killed by YARN for exceeding memory

日志报错&#xff08;WARN类型最后执行成功可以忽略&#xff09;&#xff1a; 19/04/15 12:35:37 INFO memory.MemoryStore: Will not store rdd_2_5119/04/15 12:35:37 WARN memory.MemoryStore: Not enough space to cache rdd_2_51 in memory! (computed 1109.7 MB so far)…

opencv 读取CV_16U图像 c++

cv::Mat depthImage cv::imread("../data1/depth0.png",-1);

Spark 某两个节点数据分析速度慢 - hbase数据删除(分裂) 元信息未删除导致 There is an overlap in the region chain.

基于Hbase2.0&#xff0c;Spark2.2 问题描述 执行Spark处理Hbase数据时&#xff0c;遇到某两个Excutor处理速度特别慢&#xff0c;如图 正常速度10多分钟 左右处理完成&#xff0c; 一个多小时有另外一个处理完成&#xff0c;还有一个在处理中。 分析原因 1.查看hbase数据分…

servlet中用out.print输出中文为“乱码”如何解决

最开始准备再servlet中加1.request.setCharacterEncoding("UTF-8")的作用是设置对客户端请求进行重新编码的编码。显然这个作用范围不同不可取。2.response.setCharacterEncoding("UTF-8")的作用是指定对服务器响应进行重新编码的编码。 这是在服务器端的编…

python pip清华源安装库

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple --trusted-host pypi.tuna.tsinghua.edu.cn imageio

Spark NaiveBayes Demo 朴素贝叶斯分类算法

模型sql文件 :https://pan.baidu.com/s/1hugrI9e使用数据链接 https://pan.baidu.com/s/1kWz8fNhNaiveBayes Spark Mllib训练 package com.xxx.xxx.xxximport java.io.ObjectInputStream import java.sql.{Connection, DriverManager, PreparedStatement} import java.util.{A…

TCP,UDP数据包的大小以及MTU

TCP、UDP数据包大小的确定 UDP和TCP协议利用端口号实现多项应用同时发送和接收数据。数据通过源端口发送出去&#xff0c;通过目标端口接收。有的网络应用只能使用预留或注册的静态端口&#xff1b;而另外一些网络应用则可以使用未被注册的动态端口。因为UDP和TCP报头使用两个字…

jQuery如何在线导入js包

<script src"http://libs.baidu.com/jquery/2.0.0/jquery.min.js"></script> 在<script>中加入导入的在线网络链接就ok了。 当然这种办法是你本地的包导入不进去或者是本地没有jquery的包

ubuntu 安装cudnn

tips:cudnn要与cuda对应版本下载 1.下载对应cuda版本的cudnn&#xff0c;需要注册Nvidia开发者账号。NVIDIA cuDNN下载地址:https://developer.nvidia.com/cudnn 2.解压&#xff1a;tar zxvf cudnn*.tgz 3.cuda的安装目录通过which nvcc 查看 cuda安装目录为/usr/loca/cuda…

map和hash_map

list支持快速的插入和删除&#xff0c;但是查找费时; vector支持快速的查找&#xff0c;但是插入费时。 map查找的时间复杂度是对数的&#xff0c;这几乎是最快的&#xff0c;hash也是对数的。 如果我自己写&#xff0c;我也会用二叉检索树&#xff0c;它在大部分情况下可以保…