文档章节

Kafka Shell基本命令(包括topic的增删改查)

Zero零_度
 Zero零_度
发布于 2017/09/04 19:22
字数 740
阅读 84
收藏 0

 

创建kafka topic

bin/kafka-topics.sh --zookeeper node01:2181 --create --topic t_cdr --partitions 30  --replication-factor 2

注: partitions指定topic分区数,replication-factor指定topic每个分区的副本数

  • partitions分区数:
    • partitions :分区数,控制topic将分片成多少个log。可以显示指定,如果不指定则会使用broker(server.properties)中的num.partitions配置的数量
    • 虽然增加分区数可以提供kafka集群的吞吐量、但是过多的分区数或者或是单台服务器上的分区数过多,会增加不可用及延迟的风险。因为多的分区数,意味着需要打开更多的文件句柄、增加点到点的延时、增加客户端的内存消耗。
    • 分区数也限制了consumer的并行度,即限制了并行consumer消息的线程数不能大于分区数
    • 分区数也限制了producer发送消息是指定的分区。如创建topic时分区设置为1,producer发送消息时通过自定义的分区方法指定分区为2或以上的数都会出错的;这种情况可以通过alter –partitions 来增加分区数。
  • replication-factor副本
    • replication factor 控制消息保存在几个broker(服务器)上,一般情况下等于broker的个数。
    • 如果没有在创建时显示指定或通过API向一个不存在的topic生产消息时会使用broker(server.properties)中的default.replication.factor配置的数量

 

查看所有topic列表

bin/kafka-topics.sh --zookeeper node01:2181 --list

 

查看指定topic信息

bin/kafka-topics.sh --zookeeper node01:2181 --describe --topic t_cdr

 

控制台向topic生产数据

bin/kafka-console-producer.sh --broker-list node86:9092 --topic t_cdr

 

控制台消费topic的数据

bin/kafka-console-consumer.sh  --zookeeper node01:2181  --topic t_cdr --from-beginning

 

查看topic某分区偏移量最大(小)值

bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic hive-mdatabase-hostsltable  --time -1 --broker-list node86:9092 --partitions 0

注: time为-1时表示最大值,time为-2时表示最小值

 

增加topic分区数

为topic t_cdr 增加10个分区

bin/kafka-topics.sh --zookeeper node01:2181  --alter --topic t_cdr --partitions 10

 

删除topic,慎用,只会删除zookeeper中的元数据,消息文件须手动删除

bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --zookeeper node01:2181 --topic t_cdr

 

查看topic消费进度

这个会显示出consumer group的offset情况, 必须参数为--group, 不指定--topic,默认为所有topic

Displays the: Consumer Group, Topic, Partitions, Offset, logSize, Lag, Owner for the specified set of Topics and Consumer Group

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker

required argument: [group] 
Option Description 
------ ----------- 
--broker-info Print broker info 
--group Consumer group. 
--help Print this message. 
--topic Comma-separated list of consumer 
   topics (all topics if absent). 
--zkconnect ZooKeeper connect string. (default: localhost:2181)

Example,

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group pv

Group           Topic              Pid Offset   logSize    Lag    Owner 
pv              page_visits        0   21       21         0      none 
pv              page_visits        1   19       19         0      none 
pv              page_visits        2   20       20         0      none

本文转载自:http://www.cnblogs.com/xiaodf/p/6093261.html

共有 人打赏支持
Zero零_度
粉丝 69
博文 1251
码字总数 256097
作品 0
程序员
私信 提问
mysqlbinlog入elasticsearch

环境准备: 10.99.35.214上安装elasticsearch-5.4.1.tar.gz、jdk-8u121-linux-x64.tar.gz、kibana-5.1.1-linux-x86_64.tar.gz、nginx-1.12.2.tar.gz 10.99.35.215、10.99.35.216上安装elast......

qhd2004
06/27
0
0
mysql系统数据库安装

mysql就不用谈理论了,文档写了安装过程,以及基本增删改查 1.准备工作确认没有使用RPM方式安装mysql-server、mysql2.安装mysql之前,需要安装软件开发软件包,如果安装操作系统的时候勾选过...

xuewenhcang
01/08
0
0
springboot中使用tk.mybatis代码生成器

tk.mybatis是对mybatis的进一步封装,它对基本的单表的增删改查又做了进一步封装,springboot中的使用之前已经写过了,如果没用过,请参考我之前的文章《springboot中使用PageHelper和tk.my...

梦想修补师
2017/10/28
0
0
Kafka笔记整理(三):消费形式验证与性能测试

[TOC] Kafka笔记整理(三):消费形式验证与性能测试 Kafka消费形式验证 前面的《Kafka笔记整理(一)》中有提到消费者的消费形式,说明如下: 下面就来验证Kafka的消费形式,不过需要说明的...

xpleaf
03/25
0
0
Zookeeper+Hadoop+Kafka+Scala+Spark 环境搭建(未完)

Linux环境准备 (Redhat 6.3) 环境规划 IP Hostname 192.168.0.190 master 192.168.0.191 slave1 192.168.0.192 slave2 192.168.0.193 slave3 使用"桥接"网络模式,设置固定的静态IP (或 ......

翻船全靠浪
2016/05/26
30
0

没有更多内容

加载失败,请刷新页面

加载更多

Java 源代码和 C 源代码的运行区别

与其他程序的执行方式和编译方式不同。 Java 源代码需要进行编译成字节码后在 Java 虚拟机上运行,这样 Java 程序能够保持独立性和跨平台功特性。 请参考下图。 https://www.cwiki.us/pages...

honeymose
48分钟前
3
0
Apache限定目录解析PHP,限制user_agent,PHP相关的配置

Apache限定目录解析PHP 配置前访问upload/index.php [root@test-a ~]# curl -x192.168.77.139:80 'www.test.com/upload/index.php'This is upload diretory 配置,/usr/local/apache2.4/......

野雪球
今天
5
0
java.util.Concurrent.Exchanger源码

类图 源码: package java.util.concurrent;import java.util.concurrent.atomic.AtomicInteger;import java.util.concurrent.atomic.AtomicReference;import java.util.concurrent......

狼王黄师傅
今天
6
0
Kubernetes里的secret最基本的用法

Secret解决了密码、token、密钥等敏感数据的配置问题,使用Secret可以避免把这些敏感数据以明文的形式暴露到镜像或者Pod Spec中。 Secret可以以Volume或者环境变量的方式使用。 使用如下命令...

JerryWang_SAP
昨天
5
0
2018-11-20学习笔记

1. python数据类型: 给变量赋值什么样的值,变量就是什么样的类型 给变量赋值整数,变量就是整数类型 给变量赋值字符串,变量就是字符串类型 123 和“123”一样吗? 在python中 单引号 与双...

laoba
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部