kafka命令

‘壹’ linux 怎样查看kafka的某topic数据

基于0.8.0版本。
##查看topic分布情况kafka-list-topic.sh
bin/kafka-list-topic.sh
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
（列出所有topic的分区情况）
bin/kafka-list-topic.sh
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
--topic
test
（查看test的分区情况）
其实kafka-list-topic.sh里面就一句
exec
$(dirname
$0)/kafka-run-class.sh
kafka.admin.listtopiccommand
$@
实际是通过
kafka-run-class.sh脚本执行的包kafka.admin下面的类
##创建topic
kafka-create-topic.sh
bin/kafka-create-topic.sh
--replica
2
--partition
8
--topic
test
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
创建名为test的topic，
8个分区分别存放数据，数据备份总共2份
bin/kafka-create-topic.sh
--replica
1
--partition
1
--topic
test2
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
结果
topic:
test2
partition:
0
leader:
170
replicas:
170
isr:
170
##重新分配分区kafka-reassign-partitions.sh
这个命令可以分区指定到想要的--broker-list上
bin/kafka-reassign-partitions.sh
--topics-to-move-json-file
topics-to-move.json
--broker-list
"171"
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
--execute
cat
topic-to-move.json
{"topics":
[{"topic":
"test2"}],
"version":1
}
##为topic增加
partition数目kafka-add-partitions.sh
bin/kafka-add-partitions.sh
--topic
test
--partition
2
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
（为topic
test增加2个分区）
##控制台接收消息
bin/kafka-console-consumer.sh
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
--from-beginning
--topic
test
##控制台发送消息
bin/kafka-console-procer.sh
--broker-list
192.168.197.170:9092,192.168.197.171:
9092
--topic
test
##手动均衡topic,
kafka-preferred-replica-election.sh
bin/kafka-preferred-replica-election.sh
--zookeeper
192.168.197.170:2181,192.168.197.171:2181
--path-to-json-file
preferred-click.json
cat
preferred-click.json
{
"partitions":
[
{"topic":
"click",
"partition":
0},
{"topic":
"click",
"partition":
1},
{"topic":
"click",
"partition":
2},
{"topic":
"click",
"partition":
3},
{"topic":
"click",
"partition":
4},
{"topic":
"click",
"partition":
5},
{"topic":
"click",
"partition":
6},
{"topic":
"click",
"partition":
7},
{"topic":
"play",
"partition":
0},
{"topic":
"play",
"partition":
1},
{"topic":
"play",
"partition":
2},
{"topic":
"play",
"partition":
3},
{"topic":
"play",
"partition":
4},
{"topic":
"play",
"partition":
5},
{"topic":
"play",
"partition":
6},
{"topic":
"play",
"partition":
7}
]
}
##删除topic,慎用，只会删除zookeeper中的元数据，消息文件须手动删除
bin/kafka-run-class.sh
kafka.admin.deletetopiccommand
--topic
test666
--zookeeper
192.168.197.170:2181
,192.168.197.171:2181

‘贰’ kafka为什么会自动关闭

Kafka在启动一段时间后，如果出现服务自动关闭情况，可在启动kafka的时使用守护进程模式启动，即在原启动命令中加 -daemon

kafka-server-start.sh -daemon config/server.properties &

原因参考：
kafka-run-class.sh
# Launch mode
if [ "x$DAEMON_MODE" = "xtrue" ]; then
#使用daemon执行命令
nohup $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@" > "$CONSOLE_OUTPUT_FILE" 2>&1 < /dev/null &
else
#不使用daemon执行命令
exec $JAVA $KAFKA_HEAP_OPTS $KAFKA_JVM_PERFORMANCE_OPTS $KAFKA_GC_LOG_OPTS $KAFKA_JMX_OPTS $KAFKA_LOG4J_OPTS -cp $CLASSPATH $KAFKA_OPTS "$@"
fi

‘叁’ 如何在Kafka上创建一个Topic

在运行/调试设置中，编辑配置对话框中有“Main”这个选项卡，我们可以勾选“Stop in main”这个复选框。如果选中，那么在调试一个基于main方法的Java程序时，程序会在main方法第一行位置便停止执行。

进入服务器后，找到kafka安装目录进入bin文件夹，输入命令--- 查看kafka现有主题命令:。/kafka-topics.sh --list --zookeeper zk_host:port。

‘肆’ 怎么将shell脚本的变量内容，通过kafka命令发送到kafka的topic中呢，求助

kafka $变量名

‘伍’ 如何在Kafka上对一个Topic增加partition

通过kafka-topics工具的alter命令，将TEST-TOPIC的partitions从2增加到4；
/bin/windows/kafka-topics.bat –zookeeper 127.0.0.1:2181 –alter –partitions 4–topic TEST-TOPIC

‘陆’ 如何查看kafka命令 找不到

基于0.8.0版本。

##查看topic分布情况kafka-list-topic.sh
bin/kafka-list-topic.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 （列出所有topic的分区情况）
bin/kafka-list-topic.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --topic test （查看test的分区情况）

其实kafka-list-topic.sh里面就一句
exec $(dirname $0)/kafka-run-class.sh kafka.admin.ListTopicCommand $@
实际是通过
kafka-run-class.sh脚本执行的包kafka.admin下面的类
##创建TOPIC kafka-create-topic.sh
bin/kafka-create-topic.sh --replica 2 --partition 8 --topic test --zookeeper 192.168.197.170:2181,192.168.197.171:2181
创建名为test的topic， 8个分区分别存放数据，数据备份总共2份

bin/kafka-create-topic.sh --replica 1 --partition 1 --topic test2 --zookeeper 192.168.197.170:2181,192.168.197.171:2181
结果 topic: test2 partition: 0 leader: 170 replicas: 170 isr: 170
##重新分配分区kafka-reassign-partitions.sh
这个命令可以分区指定到想要的--broker-list上
bin/kafka-reassign-partitions.sh --topics-to-move-json-file topics-to-move.json --broker-list "171" --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --execute
cat topic-to-move.json
{"topics":
[{"topic": "test2"}],
"version":1
}
##为Topic增加 partition数目kafka-add-partitions.sh
bin/kafka-add-partitions.sh --topic test --partition 2 --zookeeper 192.168.197.170:2181,192.168.197.171:2181 （为topic test增加2个分区）

##控制台接收消息
bin/kafka-console-consumer.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --from-beginning --topic test
##控制台发送消息
bin/kafka-console-procer.sh --broker-list 192.168.197.170:9092,192.168.197.171: 9092 --topic test
##手动均衡topic, kafka-preferred-replica-election.sh
bin/kafka-preferred-replica-election.sh --zookeeper 192.168.197.170:2181,192.168.197.171:2181 --path-to-json-file preferred-click.json

cat preferred-click.json
{
"partitions":
[
{"topic": "click", "partition": 0},
{"topic": "click", "partition": 1},
{"topic": "click", "partition": 2},
{"topic": "click", "partition": 3},
{"topic": "click", "partition": 4},
{"topic": "click", "partition": 5},
{"topic": "click", "partition": 6},
{"topic": "click", "partition": 7},
{"topic": "play", "partition": 0},
{"topic": "play", "partition": 1},
{"topic": "play", "partition": 2},
{"topic": "play", "partition": 3},
{"topic": "play", "partition": 4},
{"topic": "play", "partition": 5},
{"topic": "play", "partition": 6},
{"topic": "play", "partition": 7}

]
}

##删除topic,慎用，只会删除zookeeper中的元数据，消息文件须手动删除
bin/kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic test666 --zookeeper 192.168.197.170:2181 ,192.168.197.171:2181

‘柒’ 怎样查看服务器是不部署kafka

进入服务器后，找到kafka安装目录进入bin文件夹，输入命令---
查看kafka现有主题命令:。/kafka-topics.sh --list --zookeeper zk_host:port

‘捌’ kafka如何查看日志量

我们在使用kafka的过程中有时候可以需要查看我们生产的消息的各种信息，这些都是被记录在卡夫卡的log文件中的。由于log文件的特殊格式，我们是无法直接查看log文件中的信息的。本文提供一下方式查看kafka的log文件中所记录的信息。

执行命令

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

我们可以看到都需要哪些参数，根据不同的需求我们可以选择不同的参数。

执行如下命令查看某个log文件

bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files /tmp/kafka-logs/test3-0/00000000000000000000.log --print-data-log

这里 --print-data-log 是表示查看消息内容的，不加此项是查看不到详细的消息内容。如果要查看多个log文件可以用逗号分隔。

‘玖’ kafka只能用命令来创建topic么

在Kafka上创建一个Topic的步骤：进入服务器后，找到kafka安装目录进入bin文件夹，输入命令--- 查看kafka现有主题命令:。/kafka-topics.sh --list --zookeeper zk_host:port望采纳。

‘拾’ kafka在windows启动下提示：命令语法不正确

Kafka是由LinkedIn设计的一个高吞吐量、分布式、基于发布订阅模式的消息系统，使用Scala编写，它以可水平扩展、可靠性、异步通信和高吞吐率等特性而被广泛使用。目前越来越多的开源分布式处理系统都支持与Kafka集成，其中Spark Streaming作为后端流引擎配合Kafka作为前端消息系统正成为当前流处理系统的主流架构之一。然而，当下越来越多的安全漏洞、数据泄露等问题的爆发，安全正成为系统选型不得不考虑的问题，Kafka由于其安全机制的匮乏，也导致其在数据敏感行业的部署存在严重的安全隐患。本文将围绕Kafka，先介绍其整体架构和关键概念，再深入分析其架构之中存在的安全问题，最后分享下Transwarp在Kafka安全性上所做的工作及其使用方法。