Kafka

1. kafka 部署

消息中间件
mysql–>maxwell–>kafka–>ss/flink–>hbase
存储数据
缓冲上游业务高峰给下游带来的压力
作业夯住

流式平台

发布&订阅: 类似一个消息系统，读写流式的数据
处理: 可以编写可扩展的流式app，用于实时事件的响应
存储: 副本备份，容错的集群

https://docs.cloudera.com/documentation/kafka/latest.html
http://archive.cloudera.com/kafka/kafka/4/kafka-2.2.1-kafka4.1.0.tar.gz

修改kafka配置文件

[hadoop@hadoop003 config]$  vim server.properties

broker.id=0
host.name=hadoop003
port=9092
log.dirs=/home/hadoop/tmp/kafka-logs
zookeeper.connect=hadoop003:2181,hadoop004:2181,hadoop005:2181/kafka

[hadoop@hadoop004 config]$  vim server.properties
broker.id=1
host.name=hadoop004
port=9092
log.dirs=/home/hadoop/tmp/kafka-logs
zookeeper.connect=hadoop003:2181, hadoop004:2181, hadoop005:2181/kafka

[hadoop@hadoop005 config]$  vim server.properties

broker.id=2
host.name=hadoop005
port=9092
log.dirs=/home/hadoop/tmp/kafka-logs
zookeeper.connect=hadoop003:2181,hadoop004:2181,hadoop005:2181/kafka


[hadoop@hadoop003 config]$ vim zookeeper.properties
dataDir=/home/hadoop/tmp/zookeeper


[hadoop@hadoop003 bin]$ vim kafka-run-class.sh

if [ -z "$JAVA_HOME" ]; then
  JAVA="/usr/java/jdk"
else
  JAVA="$JAVA_HOME/bin/java"
fi


[hadoop@hadoop003 bin]$ vim kafka-server-stop.sh
SIGNAL=${SIGNAL:-TERM}
PIDS=$(jps -lm | grep -i 'kafka.Kafka'| awk '{print $1}')

[hadoop@hadoop003 app]$ scp -r kafka_2.11-2.2.1-kafka-4.1.0/ hadoop005:/home/hadoop/app/
[hadoop@hadoop003 app]$ scp -r kafka_2.11-2.2.1-kafka-4.1.0/ hadoop004:/home/hadoop/app/

启动集群

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

查看jps -m

查看topic

kafka-topics.sh --list \
--zookeeper  hadoop003:2181,hadoop004:2181,hadoop005:2181/kafka

创建topic

kafka-topics.sh --create \
--zookeeper  hadoop003:2181,hadoop004:2181,hadoop005:2181/kafka \
--partitions 3 \
--replication-factor 3 \
--topic test

描述

kafka-topics.sh --describe \
--zookeeper  hadoop003:2181,hadoop004:2181,hadoop005:2181/kafka \
--topic test

6.console案例

kafka-console-producer.sh \
--broker-list hadoop003:9092,hadoop004:9092,hadoop005:9092 \
--topic test

kafka-console-consumer.sh \
--bootstrap-server hadoop003:9092,hadoop004:9092,hadoop005:9092 \
--topic test \
--from-beginning

2. kafka 启动脚本编写

#!/bin/bash
case $1 in
"start"){
    
        for i in hadoop003 hadoop004 hadoop005
        do
                echo "**********$i kafka start **********"
                ssh hadoop@$i "kafka-server-start.sh -daemon /home/hadoop/app/kafka/config/server.properties"
                echo "$i kafka is running"
        done
};;

"stop"){
    
        for i in hadoop003 hadoop004 hadoop005
        do
                echo "**********$i kafka stop**********"
                ssh hadoop@$i "/home/hadoop/app/kafka/bin/kafka-server-stop.sh /home/hadoop/app/kafka/config/server.properties"
        done
};;
esac

ln -s /usr/mybin/kafka.sh /usr/local/bin/kafka

3. kafka Topic

1. topic 数据主题，数据记录发布的地方 用来区分业务系统

2. partition: topic物理上的分组
   一个topic可以分为多个partition
   每个partition都是一个[有序]的队列
   其实就是一个文件夹而已

命名规则—>topic名称-partition序号
如何保证消费有序： 一个partition
producer_parititon_by =table maxwell 自己封装

4. kafka 分区策略

DefaultPartitioner.ava

如何分区：
int parititon
=> send(k,v)

给它设置单分区有序
设置K的值 （通过hash算法设置同个维度的数据放入同个分区）保证他们有序

ack= all 最可靠 性能低
ack=1 每个分区的老大写成功 性能高 保障级别低

retries 100
max.in.flight.requests.per.connection 5 -> 1
如果不设置1，会改变消息的顺序