大数据日志分析系统-缓存组件kafka

2018年12月18日 770 字大概 3 分钟

kafka简介

是一种高吞吐量的分布式发布订阅消息系统，当数据量不稳定，数据量大的时候想到它就对了。

zookeeper简介

是一个分布式的，开放源码的分布式应用程序协调服务,很多地方用到, 最常见的是为集群提供基础的、高可用HA(High Availability)服务
是kafka集群的基础依赖，同时也是hadoop系列中实现HA的基础组件。
实现HDFS的NamaNode和YARN的ResourceManager的HA,Spark实现HA，
HBase主要用ZooKeeper来实现HMaster选举与主备切换、系统容错、RootRegion管理、Region状态管理和分布式SplitWAL任务管理等。

//===========================================================

两种集群的配置相差不大

zookeeper集群配置：

ubuntu@sp1:~/kafka/config$ cat zookeeper.properties | grep -v '#'
dataDir=/mnt/data3/zk
clientPort=2181
tickTime=2000
initLimit=7
syncLimit=4
server.1=sp1:2888:3888
server.2=sp2:2888:3888
server.3=sp3:2888:3888
server.4=sp4:2888:3888
server.5=sp5:2888:3888
server.6=sp6:2888:3888
server.7=sp7:2888:3888

强制杀死命令

kafka 集群配置：

ubuntu@sp1:~/kafka/config$ cat consumer.properties | grep -v '#'
zookeeper.connect=es1:2181,es2:2181,es3:2181,es4:2181,es5:2181
zookeeper.connection.timeout.ms=6000
group.id=test-consumer-group

ubuntu@sp1:~/kafka/config$ cat server.properties | grep -v '#'
broker.id=1
port=9092
advertised.host.name=sp1
listeners=PLAINTEXT://sp1-host-ip:9092  
num.network.threads=10
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/mnt/data3/kafka-logs
num.partitions=20
num.recovery.threads.per.data.dir=5
log.retention.hours=8
log.retention.bytes=53687091200
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
log.cleaner.enable=false
zookeeper.connect=sp1:2181,sp2:2181,sp3:2181,sp4:2181,sp5:2181,sp6:2181,sp7:2181
zookeeper.connection.timeout.ms=6000
delete.topic.enable=true
auto.leader.rebalance.enable=true
num.replica.fetchers=5

启动命令：

1.zookeeper启动

sp1 - sp7在Ubuntu用户下执行

/home/ubuntu/zookeeper-3.4.7/bin/zkServer.sh start

jps检查存在

QuorumPeerMain

状态查看

/home/ubuntu/zookeeper-3.4.7/bin/zkServer.sh status

查看为一主动多从启动正常

文件查看

/home/ubuntu/zookeeper-3.4.7/bin/zkCli.sh -server localhost:2181

进入客户端 ls / 等命令查看

2.kafka启动

sp1到sp7执行

nohup /home/ubuntu/kafka/bin/kafka-server-start.sh /home/ubuntu/kafka/config/server.properties 2>&1 > /home/ubuntu/kafka/logs/kafka.log &

当然也可以这样远程执行

ssh sp1 nohup /home/ubuntu/kafka/bin/kafka-server-start.sh /home/ubuntu/kafka/config/server.properties 2>&1 > /home/ubuntu/kafka/logs/kafka.log &

检查

~/kafka/bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker –broker-info –group kafka_es –topic kafka_es –zookeeper localhost:2181

就可以看机器是否正常了

注意：

很可能出现kafka个别分片堆积问题：例如现在kafka有35个分片，只有2个分片产生堆积，通过命令

~/kafka/bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --broker-info --group kafka_es --topic kafka_es --zookeeper localhost:2181

查看，我遇见过是消费线程小于分片数量（现象是用命令查看是不同的分片Pid 拥有相同的 Owner），这个时候增加消费者进程即可（我个人认为数据量大时消费者要多于分片数量这样不容易出现挂掉一两个消费者出现分片被堆积的情况）　