kafka环境搭建与实战(1)安装kafka

kafka环境搭建与实战(1)安装kafka           http://zilongzilong.iteye.com/blog/2267913
kafka环境搭建与实战(2)kafka API实战       http://zilongzilong.iteye.com/blog/2267924

1.环境介绍

      3台机器IP为:

                      192.168.88.20(hostname=kafka0)

                      192.168.88.21(hostname=kafka1)

                      192.168.88.22(hostname=kafka2) 

2.zookeeper3.4.6集群安装

1) 下载解压zookeeper3.4.6

        下载zookeeper-3.4.6.tar.gz到/opt

        解压tar -zxvf zookeeper-3.4.6.tar.gz

        注意：上面3台机器都要安装

2) 配置/etc/hosts

 

# Do not remove the following line, or various programs
# that require network functionality will fail.
127.0.0.1               localhost.localdomain localhost
::1             localhost6.localdomain6 localhost6
192.168.88.22 kafka2
192.168.88.21 kafka1
192.168.88.20 kafka0

    注意：上面3台机器都要配置

 

3) 创建zookeeper数据文件

 

sudo rm -r /home/hadoop/zookeeper
cd  /home/hadoop
mkdir zookeeper

 

    注意：上面3台机器都要创建，我环境中是独立创建了hadoop用户，单独用于启动大数据相关服务，这里也是在hadoop用户目录下创建zookeeper配置文件夹

4) 创建zookeeper数据文件

       将/opt/zookeeper-3.3.4/conf目录下面的 zoo_sample.cfg修改为zoo.cfg，配置文件内容如下所示:

 

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/home/hadoop/zookeeper
clientPort=2181
server.1=kafka0:2888:3888
server.2=kafka1:2888:3888
server.3=kafka2:2888:3888
#数据文件保存最近的3个快照，默认是都保存，时间长的话会占用很大磁盘空间
autopurge.snapRetainCount=3
#单位为小时，每小时清理一次快照数据
autopurge.purgeInterval=1

 

 

     注意：上面3台机器都要配置

 

5) 创建zookeeper集群节点标识文件myid

 

#在机器kafka0下/home/hadoop/zookeeper/创建文件myid，写入数字1
ssh kafka0
echo "1" > /home/hadoop/zookeeper/myid  
#在机器kafka1下/home/hadoop/zookeeper/创建文件myid，写入数字1
ssh kafka1
echo "2" > /home/hadoop/zookeeper/myid  
#在机器kafka2下/home/hadoop/zookeeper/创建文件myid，写入数字1
ssh kafka2
echo "3" > /home/hadoop/zookeeper/myid  

      注意：上面3台机器都要配置,各自标识必须不相同，并且为整数

 

6) 启动ZooKeeper集群

 

cd /opt/zookeeper-3.4.6
bin/zkServer.sh start  

 

7) 查看单机ZooKeeper是leader还是follower

 

cd /opt/zookeeper-3.4.6
bin/zkServer.sh status 

 

8) 停止ZooKeeper集群

 

cd /opt/zookeeper-3.4.6
bin/zkServer.sh stop 

 

3.kafka_2.10-0.9.0.0集群安装

1) 下载解压kafka_2.10-0.9.0.0.tgz

 

下载kafka_2.10-0.9.0.0.tgz到/opt
解压tar -zxvf kafka_2.10-0.9.0.0.tgz

 注意：上面3台机器都要安装

 

2) 配置kafka各集群节点

修改/opt/kafka_2.10-0.9.0.0/config/server.properties,修改内容如下：

 

#机器kafka0
#每个机器上应该不同，且为数字
broker.id=1
#端口默认保持不变
port=9092  
#本机器的IP地址
host.name=kafka0
#zookeeper的链接地址
zookeeper.connect=kafka0:2181,kafka1:2181,kafka2:2181 
#kafka中每个TOPIC中数据最多只保留当前时间往前推8小时的数据
log.retention.hours=8
#kafka中每个TOPIC切分为4份，分布式存贮到集群节点，这个是默认值，可以创建TOPIC时单独指定
num.partitions=4
#kafka中每个TOPIC数据存储的目录位置
log.dirs=/tmp/kafka-logs
#启用对于TOPIC的删除功能
delete.topic.enable = true

 

#机器kafka2
#每个机器上应该不同，且为数字
broker.id=2
#端口默认保持不变
port=9092  
#本机器的IP地址
host.name=kafka2
#zookeeper的链接地址
zookeeper.connect=kafka0:2181,kafka1:2181,kafka2:2181 
#kafka中每个TOPIC中数据最多只保留当前时间往前推8小时的数据
log.retention.hours=8
#kafka中每个TOPIC切分为4份，分布式存贮到集群节点，这个是默认值，可以创建TOPIC时单独指定
num.partitions=4
#kafka中每个TOPIC数据存储的目录位置
log.dirs=/tmp/kafka-logs
#启用对于TOPIC的删除功能
delete.topic.enable = true

 

#机器kafka1
#每个机器上应该不同，且为数字
broker.id=3
#端口默认保持不变
port=9092  
#本机器的IP地址
host.name=kafka1
#zookeeper的链接地址
zookeeper.connect=kafka0:2181,kafka1:2181,kafka2:2181 
#kafka中每个TOPIC中数据最多只保留当前时间往前推8小时的数据
log.retention.hours=8
#kafka中每个TOPIC切分为4份，分布式存贮到集群节点，这个是默认值，可以创建TOPIC时单独指定
num.partitions=4
#kafka中每个TOPIC数据存储的目录位置
log.dirs=/tmp/kafka-logs
#启用对于TOPIC的删除功能
delete.topic.enable = true

 

 

3) 启动kafka，关闭kafka

 

#后台进程启动kafka
/opt/kafka_2.10-0.9.0.0/bin/kafka-server-start.sh /opt/kafka_2.10-0.9.0.0/config/server.properties &
#console启动kafka
/opt/kafka_2.10-0.9.0.0/bin/kafka-server-start.sh /opt/kafka_2.10-0.9.0.0/config/server.properties
#关闭kafka，先通过jps查看进程号后kill - 9 进程号

 

 

4) 在kafka中创建名为“test”的topic，该topic切分为4份，每一份备份数为3

 

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 4 --topic  test

 

 

5) 在kafka中查看名为“test”的topic

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-topics.sh --describe --zookeeper kafka0:2181,kafka1:2181,kafka2:2181 --topic test

 

6) 在kafka中列出所有topic

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-topics.sh --list --zookeeper kafka0:2181,kafka1:2181,kafka2:2181

 

7) 在kafka中删除名为“test”的topic

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-topics.sh --delete --zookeeper kafka0:2181,kafka1:2181,kafka2:2181 --topic test

 

8) 在kafka中通过console启动一个消费者用来消费来自topic=test里面的数据

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-console-consumer.sh --zookeeper kafka0:2181,kafka1:2181,kafka2:2181 --topic test --from-beginning

 

9) 在kafka中通过console启动一个生产者用来产生数据放入topic=test

cd /opt/kafka_2.10-0.9.0.0
bin/kafka-console-producer.sh --broker-list kafka0:9092,kafka1:9092,kafka2:9092 --topic test