3.kafka 基本配置

1.主題管理

kafka-topics.sh工具腳本用於對主題操做，如建立、刪除、修改、分區數、副本數及主題級別的配置。

1.1建立名爲kafka-test主題，有2個副本，3個分區

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test

輸出：

Created topic "kafka-test".

####

--zookeeper 參數必傳，用於與zookeeper鏈接

--partitions 參數必傳，用於設置分區數

--replication-factor 參數必傳，用於設置副本數

####

 

1.2進入log目錄

[hadoop@h201 kafka_2.12-0.10.2.1]$ cd kafkalogs/

[hadoop@h201 kafkalogs]$ ls

kafka-test-1kafka-test-0

##副本分佈到不一樣的節點上###

h202上

[hadoop@h202 kafkalogs]$ ls

kafka-test-2kafka-test-1

 

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh 

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

結果：

[0, 1, 2]

[zk: localhost:2181(CONNECTED) 2] get /brokers/topics/kafka-test

 

1.3 建立主題時，能夠經過config參數設置主題級別以覆蓋默認配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test1 --config max.message.bytes=404800

 

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh 

[zk: localhost:2181(CONNECTED) 1] get /config/topics/kafka-test1

結果：

{"version":1,"config":{"max.message.bytes":"404800"}}

 

2.刪除主題

若想完全刪除，啓動時加載server.properties文件中配置delete.topic.enable=true,默認爲false

若是沒有設置爲true，只是標記kafka-test1爲刪除狀態，當設置爲true時刪除文件目錄及元數據。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --delete --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test1

 

3.查看主題

3.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --list --zookeeper h201:2181,h202:2181,h203:2181

3.2查看詳細描述信息

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181

 

3.3 查看處於正在同步主題

能夠看到正在處於「under replicated」狀態分區，該狀態可能爲正在同步或同步異常。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --under-replicated-partitions

 

3.4 查看沒有leader的分區

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --unavailable-partitions

 

3.5 查看主題覆蓋的配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --topics-with-overrides

 

4.修改主題

4.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --config segment.bytes=209715200

##段大小 修改成200m##

 

4.2 刪除修改

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --delete-config segment.bytes

 

4.3 增長分區

Kafka不支持減小分區操做，只能增長分區操做

主題kafka-test當前分區數爲3個，調整爲5個

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --partitions 5

 

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

結果：

[0, 1, 2, 3, 4]

5.生產者操做

使用kafka-console-producer.sh腳本

參數producer.config：用於加載一個生產者配置文件

參數producer-property：用於設置參數將會覆蓋所加載的配置文件中的參數

參數property：用於設置消息消費者相關配置

 

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

必傳參數--broker-list，--topic

測試：（顯示主題下各個分區的偏移量）

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test --time -1

##kafka-run-class.sh爲kafka自帶測試工具###

 

6.消費者操做

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

 

消費者建立時會向zookeeper中註冊元數據信息 /consumers/${group.id}

---------------------測試-------------------------

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

添加測試消息內容

hello kafka

 

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

可以接收到 「hello kafka」消息內容

1.  主題管理

kafka-topics.sh工具腳本用於對主題操做，如建立、刪除、修改、分區數、副本數及主題級別的配置。

1.1  建立名爲kafka-test主題，有2個副本，3個分區

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test

輸出：

Created topic "kafka-test".

####

--zookeeper 參數必傳，用於與zookeeper鏈接

--partitions 參數必傳，用於設置分區數

--replication-factor 參數必傳，用於設置副本數

####

1.2進入log目錄

[hadoop@h201 kafka_2.12-0.10.2.1]$ cd kafkalogs/

[hadoop@h201 kafkalogs]$ ls

kafka-test-1kafka-test-0

##副本分佈到不一樣的節點上###

h202上

[hadoop@h202 kafkalogs]$ ls

kafka-test-2kafka-test-1

 

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

結果：

[0, 1, 2]

[zk: localhost:2181(CONNECTED) 2] get /brokers/topics/kafka-test

 

1.3 建立主題時，能夠經過config參數設置主題級別以覆蓋默認配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --create --zookeeper h201:2181,h202:2181,h203:2181 --replication-factor 2 --partitions 3 --topic kafka-test1 --config max.message.bytes=404800

 

[hadoop@h201 zookeeper-3.4.5-cdh5.5.2]$ bin/zkCli.sh

[zk: localhost:2181(CONNECTED) 1] get /config/topics/kafka-test1

結果：

{"version":1,"config":{"max.message.bytes":"404800"}}

 

2.  刪除主題

若想完全刪除，啓動時加載server.properties文件中配置delete.topic.enable=true,默認爲false

若是沒有設置爲true，只是標記kafka-test1爲刪除狀態，當設置爲true時刪除文件目錄及元數據。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --delete --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test1

 

3.  查看主題

3.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --list --zookeeper h201:2181,h202:2181,h203:2181

3.2查看詳細描述信息

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181

 

3.3 查看處於正在同步主題

能夠看到正在處於「under replicated」狀態分區，該狀態可能爲正在同步或同步異常。

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --under-replicated-partitions

 

3.4 查看沒有leader的分區

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --unavailable-partitions

 

3.5 查看主題覆蓋的配置

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --describe --zookeeper h201:2181,h202:2181,h203:2181 --topics-with-overrides

 

4.  修改主題

4.1

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --config segment.bytes=209715200

##段大小 修改成200m##

 

4.2 刪除修改

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --delete-config segment.bytes

 

4.3 增長分區

Kafka不支持減小分區操做，只能增長分區操做

主題kafka-test當前分區數爲3個，調整爲5個

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-topics.sh --alter --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --partitions 5

 

[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/kafka-test/partitions

結果：

[0, 1, 2, 3, 4]

5.  生產者操做

使用kafka-console-producer.sh腳本

參數producer.config：用於加載一個生產者配置文件

參數producer-property：用於設置參數將會覆蓋所加載的配置文件中的參數

參數property：用於設置消息消費者相關配置

 

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

必傳參數--broker-list，--topic

測試：（顯示主題下各個分區的偏移量）

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test --time -1

##kafka-run-class.sh爲kafka自帶測試工具###

 

6.  消費者操做

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

 

消費者建立時會向zookeeper中註冊元數據信息 /consumers/${group.id}

---------------------測試-------------------------

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-producer.sh --broker-list h201:9092,h202:9092,h203:9092 --topic kafka-test

添加測試消息內容

hello kafka

 

[hadoop@h201 kafka_2.12-0.10.2.1]$ bin/kafka-console-consumer.sh --zookeeper h201:2181,h202:2181,h203:2181 --topic kafka-test --consumer-property group.id=consumer-g1  --from-beginning

可以接收到 「hello kafka」消息內容