Kafka分布式集群部署

 

---

　　Kafka是由LinkedIn开发的一个分布式的消息系统，使用Scala编写，它以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如Cloudera、Apache Storm、Spark都支持与Kafka集成。

1.下载Kafka并安装

• kafka是一个消息系统。
• kafka对流数据可以高效的实时处理。
• 分布式集群的环境下能够保证数据的安全。

　1）下载Apache版本的Kafka。

　2）下载Cloudera版本的Kafka。

　3）这里选择下载Apache版本的kafka_2.11-0.8.2.1.tgz ，然后上传至bigdata-pro01.kfk.com节点/opt/softwares/目录下

[kfk@bigdata-pro01 ~]$ cd /opt/softwares/

[kfk@bigdata-pro01 softwares]$ rz

　4）解压Kafka

[kfk@bigdata-pro01 softwares]$ tar -zxf kafka_2.11-0.8.2.1.tgz  -C /opt/modules/

[kfk@bigdata-pro01 softwares]$ cd ../modules/

[kfk@bigdata-pro01 modules]$ ll

total 20

drwxr-xr-x  11 kfk kfk 4096 Oct 22 12:09 hadoop-2.6.0

drwxr-xr-x  23 kfk kfk 4096 Oct 23 10:04 hbase-0.98.6-cdh5.3.0

drwxr-xr-x.  8 kfk kfk 4096 Aug  5  2015 jdk1.8.0_60

drwxr-xr-x   5 kfk kfk 4096 Feb 27  2015 kafka_2.11-0.8.2.1

drwxr-xr-x  15 kfk kfk 4096 Oct 22 15:56 zookeeper-3.4.5-cdh5.10.0

2.Kafka集群配置

　1）配置server.properties文件

vi kafka_2.11-0.8.2.1

#节点唯一标识

broker.id=0

#默认端口号

port=9092

#主机名绑定

host.name=bigdata-pro01.kfk.com

#Kafka数据目录

log.dirs=/opt/modules/kafka_2.11-0.8.2.1/tmp/kafka-logs

#配置Zookeeper

zookeeper.connect=bigdata-pro01.kfk.com:2181,bigdata-pro02.kfk.com:2181,bigdata-pro03.kfk.com:2181

　2）配置zookeeper.properties文件

　　#配置hostname

 

　　#Zookeeper的数据存储路径

[kfk@bigdata-pro01 modules]$ cd kafka_2.11-0.8.2.1/

[kfk@bigdata-pro01 kafka_2.11-0.8.2.1]$ mkdir kafka-logs

[kfk@bigdata-pro01 kafka_2.11-0.8.2.1]$ cd kafka-logs/

[kfk@bigdata-pro01 kafka-logs]$ pwd

/opt/modules/kafka_2.11-0.8.2.1/kafka-logs

 

　3）配置producer.properties文件

　　#配置Kafka集群地址

metadata.broker.list=bigdata-pro01.kfk.com:9092

 

　4）Kafka分发到其他节点

scp -r kafka_2.11-0.8.2.1 bigdata-pro02.kfk.com:/opt/modules/

scp -r kafka_2.11-0.8.2.1 bigdata-pro03.kfk.com:/opt/modules/

　5）修改另外两个节点的server.properties

#bigdata-pro02.kfk.com节点

broker.id=1

host.name=bigdata-pro02.kfk.com

#bigdata-pro03.kfk.com节点
 
broker.id=2

host.name=bigdata-pro03.kfk.com

 

3.启动Kafka集群并进行测试

　1）各个节点启动Zookeeper集群

bin/zkServer.sh start

　

　2）启动Kafka集群

bin/kafka-server-start.sh config/server.properties

 

　3）创建topic

[kfk@bigdata-pro02 kafka_2.11-0.8.2.1]$ bin/kafka-topics.sh --zookeeper bigdata-pro01.kfk.com:2181,bigdata-pro02.kfk.com:2181,bigdata-pro03.kfk.com:2181 --create --topic test --replication-factor 1 --partitions 1

Created topic "test".

 

　4）查看topic列表

[kfk@bigdata-pro03 kafka_2.11-0.8.2.1]$ cd ../zookeeper-3.4.5-cdh5.10.0/

[kfk@bigdata-pro03 zookeeper-3.4.5-cdh5.10.0]$ bin/zkCli.sh

Connecting to localhost:2181

2018-10-24 11:25:52,530 [myid:] - INFO  [main:Environment@100] - Client environment:zookeeper.version=3.4.5-cdh5.10.0--1, built on 01/20/2017 20:07 GMT

2018-10-24 11:25:52,536 [myid:] - INFO  [main:Environment@100] - Client environment:host.name=bigdata-pro03.kfk.com

2018-10-24 11:25:52,536 [myid:] - INFO  [main:Environment@100] - Client environment:java.version=1.8.0_60

2018-10-24 11:25:52,540 [myid:] - INFO  [main:Environment@100] - Client environment:java.vendor=Oracle Corporation

2018-10-24 11:25:52,540 [myid:] - INFO  [main:Environment@100] - Client environment:java.home=/opt/modules/jdk1.8.0_60/jre

2018-10-24 11:25:52,540 [myid:] - INFO  [main:Environment@100] - Client environment:java.class.path=/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../build/classes:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../build/lib/*.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/zookeeper-3.4.5-cdh5.10.0.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/slf4j-log4j12-1.7.5.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/slf4j-api-1.7.5.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/netty-3.10.5.Final.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/log4j-1.2.16.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../share/zookeeper/jline-2.11.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../src/java/lib/*.jar:/opt/modules/zookeeper-3.4.5-cdh5.10.0/bin/../conf:

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:java.library.path=/usr/java/packages/lib/amd64:/usr/lib64:/lib64:/lib:/usr/lib

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:java.io.tmpdir=/tmp

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:java.compiler=<NA>

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:os.name=Linux

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:os.arch=amd64

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:os.version=2.6.32-431.el6.x86_64

2018-10-24 11:25:52,541 [myid:] - INFO  [main:Environment@100] - Client environment:user.name=kfk

2018-10-24 11:25:52,542 [myid:] - INFO  [main:Environment@100] - Client environment:user.home=/home/kfk

2018-10-24 11:25:52,542 [myid:] - INFO  [main:Environment@100] - Client environment:user.dir=/opt/modules/zookeeper-3.4.5-cdh5.10.0

2018-10-24 11:25:52,545 [myid:] - INFO  [main:ZooKeeper@438] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@1a86f2f1

Welcome to ZooKeeper!

2018-10-24 11:25:52,609 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@975] - Opening socket connection to server localhost/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)

JLine support is enabled

2018-10-24 11:25:52,759 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@852] - Socket connection established, initiating session, client: /0:0:0:0:0:0:0:1:41773, server: localhost/0:0:0:0:0:0:0:1:2181

2018-10-24 11:25:52,784 [myid:] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1235] - Session establishment complete on server localhost/0:0:0:0:0:0:0:1:2181, sessionid = 0x366a40512f10002, negotiated timeout = 30000

 

WATCHER::

 

WatchedEvent state:SyncConnected type:None path:null

[zk: localhost:2181(CONNECTED) 0] ls

[zk: localhost:2181(CONNECTED) 1] ls /

[controller_epoch, controller, brokers, zookeeper, yarn-leader-election, hadoop-ha, rmstore, admin, consumers, config, hbase]

[zk: localhost:2181(CONNECTED) 2] ls /brokers

[ids, topics]

[zk: localhost:2181(CONNECTED) 3] ls /brokers/topics

[test]

 

　5）生产者生成数据以及消费者消费数据

　　创建生产者

[kfk@bigdata-pro02 kafka_2.11-0.8.2.1]$ bin/kafka-console-producer.sh --broker-list bigdata-pro01.kfk.com:9092,bigdata-pro02.kfk.com:9092,bigdata-pro03.kfk.com:9092 --topic test

[2018-10-24 11:28:52,335] WARN Property topic is not valid (kafka.utils.VerifiableProperties)

　　创建消费者

[kfk@bigdata-pro02 kafka_2.11-0.8.2.1]$ bin/kafka-console-consumer.sh --zookeeper bigdata-pro01.kfk.com:2181,bigdata-pro02.kfk.com:2181,bigdata-pro03.kfk.com:2181 --from-beginning --topic test

　　在生产者这边输入数据：

 

　　消费者端便可以实时的获取数据：

---

以上就是博主为大家介绍的这一板块的主要内容，这都是博主自己的学习过程，希望能给大家带来一定的指导作用，有用的还望大家点个支持，如果对你没用也望包涵，有错误烦请指出。如有期待可关注博主以第一时间获取更新哦，谢谢！同时也欢迎转载，但必须在博文明显位置标注原文地址，解释权归博主所有！