转载请注明出处:http://www.cnblogs.com/xiaodf/

创建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

以上图中参数含义解释如下:

topic:创建时topic名称

pid:分区编号

offset:表示该parition已经消费了多少条message

logSize:表示该partition已经写了多少条message

Lag:表示有多少条message没有被消费。

Owner:表示消费者

细看kafka-run-class.sh脚本,它是调用 了ConsumerOffsetChecker的main方法,所以,我们也可以通过java代码来访问scala的ConsumerOffsetChecker类,代码如下:

import kafka.tools.ConsumerOffsetChecker;

/**
* kafka自带很多工具类,其中ConsumerOffsetChecker能查看到消费者消费的情况,
* ConsumerOffsetChecker只是将信息打印到标准的输出流中
*
*/
public class RunClass {
public static void main(String[] args) {
//group-1是消费者的group名称,可以在zk中
String[] arr = new String[]{"--zookeeper=192.168.199.129:2181,192.168.199.130:2181,192.168.199.131:2181/kafka","--group=group-1"};
ConsumerOffsetChecker.main(arr);
}
}

更多文章

MapReduce Shuffle 和 Spark Shuffle 原理概述

知识图谱在大数据中的应用

基于 Apache Flink 和规则引擎的实时风控解决方案

04-20 11:35