搭建5节点hadoop+zookeeper+hbase+spark+kafka+storm(5):Kafka

本篇介绍Kafka集群安装。
版本:java 1.8,Hadoop 2.7.3,HBase 1.2.5,Zookeeper 3.4.10,Spark 2.1.1,Scala 2.12.2,Kafka 0.10.2.1,Storm 1.1.0。
以下操作都是以root用户进行,如需用户组和权限设定需自行配置。


1. 服务器信息及节点分配

服务器信息:

No Hostname 内网IP 外网IP OS
1 node01 192.168.5.11 192.168.205.50 Centos7
2 node02 192.168.5.12 192.168.205.62 Centos7
3 node03 192.168.5.13 192.168.205.63 Centos7
4 node04 192.168.5.14 192.168.205.70 Centos7
5 node05 192.168.5.15 192.168.205.102 Centos7

节点分配:

节点 node01 node02 node03 node04 node05
namenode YES YES NO NO NO
datanode NO NO YES YES YES
journalnode YES YES YES YES YES
zookeeper YES YES YES YES YES
hbase 主Master 备份Master RegionServer RegionServer RegionServer
spark YES YES YES YES YES
kafka YES YES YES YES YES
storm YES YES YES YES YES

2. 安装Kafka

2.1 下载安装包

1
wget http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.10.2.1/kafka_2.12-0.10.2.1.tgz 

2.2 创建模板配置文件

需要修改的配置文件为server.properties。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

# Switch to enable topic deletion or not, default value is false
#delete.topic.enable=true

############################# Socket Server Settings #############################

# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://node01:9092

# Hostname and port the broker will advertise to producers and consumers. If not set,
# it uses the value for "listeners" if configured. Otherwise, it will use the value
# returned from java.net.InetAddress.getCanonicalHostName().
#advertised.listeners=PLAINTEXT://node01:9092

# Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
#listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

# The number of threads handling network requests
num.network.threads=3

# The number of threads doing disk I/O
num.io.threads=8

# The send buffer (SO_SNDBUF) used by the socket server
socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server
socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)
socket.request.max.bytes=104857600


############################# Log Basics #############################

# A comma seperated list of directories under which to store log files
log.dirs=/tmp/kafka-logs

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=1

# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
# This value is recommended to be increased for installations with data dirs located in RAID array.
num.recovery.threads.per.data.dir=1

############################# Log Flush Policy #############################

# Messages are immediately written to the filesystem but by default we only fsync() to sync
# the OS cache lazily. The following configurations control the flush of data to disk.
# There are a few important trade-offs here:
# 1. Durability: Unflushed data may be lost if you are not using replication.
# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
# The settings below allow one to configure the flush policy to flush data after a period of time or
# every N messages (or both). This can be done globally and overridden on a per-topic basis.

# The number of messages to accept before forcing a flush of data to disk
#log.flush.interval.messages=10000

# The maximum amount of time a message can sit in a log before we force a flush
#log.flush.interval.ms=1000

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes. Functions independently of log.retention.hours.
#log.retention.bytes=1073741824

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=node01:2181,node02:2181,node03:2181,node04:2181,node05:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

其中:

  • broker.id:broker的id,必需是一个全局(集群)唯一的整数值,即集群中每个kafka server的配置不能相同。此处按node01-1/node02-2/node03-3/node04-4/node05-5来配置。
  • listeners:socket监听的地址,格式为listeners = security_protocol://host_name:port,端口默认为9092。
  • log.dirs:日志保存目录
  • zookeeper.connect:zookeeper连接地址

2.3 安装部署

写一个简易部署脚本:

kafka.sh
1
2
3
4
5
6
7
8
9
#/bin/bash

tar -xvf ./kafka_2.12-0.10.2.1.tgz
mv ./kafka_2.12-0.10.2.1 /usr/kafka

cp -rf ./server.properties /usr/kafka/config/server.properties

echo "export KAFKA_HOME=/usr/kafka" >> /root/.bash_profile
echo "export PATH=\$KAFKA_HOME/bin:\$PATH" >> /root/.bash_profile

修改权限并执行,执行时确保kafka_2.12-0.10.2.1.tgz,server.properties,kafka.sh在同一目录下:
1
2
3
chmod +x ./kafka.sh
./kafka.sh
source /root/.bash_profile

2.4 部署所有节点

使用sftp将kafka_2.12-0.10.2.1.tgz,server.properties,kafka.sh传到其他所有节点上并按照以上步骤进行部署,注意根据具体节点修改server.properties中的broker.id和listeners。


3. Kafka使用

3.1 启动Kafka集群

在所有节点上执行命令:

1
kafka-server-start.sh -daemon /usr/kafka/config/server.properties

3.2 创建Topic

1
kafka-topics.sh --zookeeper node01:2181 --create --replication-factor 1 --partitions 1 --topic test

3.3 查看Topic

1
kafka-topics.sh --zookeeper node01:2181 --list

3.4 创建Producer

1
kafka-console-producer.sh --broker-list node01:9092 --topic test

3.5 创建Consumer

1
kafka-console-consumer.sh --bootstrap-server node01:9092 --topic test --from-beginning

3.6 删除Topic

1
kafka-topics.sh --zookeeper node01:2181 --delete --topic test

参照资料:
Kafka0.10.2.0分布式集群安装
Kafka学习总结(五)——Kafka集群搭建
Kafka【第一篇】Kafka集群搭建
Kafka学习(三):Kafka的内部机制深入(持久化,分布式,通讯协议)