Kafka官网 kafka.apache.org
1.官网的介绍
消息中间件 承上启下 缓冲 稳稳的消费
flume-->kafka-->ss/flink
distributed event streaming platform
2.集群部署
条件:
3台 按量付费机器
部署jdk zookeeper
kafka不再需要zk，建议新的功能 生产还是等等
90% 消息中间件
10% 流式平台开发 不适合市场
kafka安装包下载
http://archive.cloudera.com/kafka/kafka/4/
kafka-2.2.1-kafka4.1.0.tar.gz
安装文档
[hadoop@ruozedata001 config]$ pwd
/home/hadoop/app/kafka/config
[hadoop@ruozedata001 config]$ vi server.properties
broker.id=0
host.name=ruozedata001
port=9092
log.dirs=/home/hadoop/tmp/kafka-logs
小尾巴 ： /kafka 这个很重要后期升级删除zookeeper信息
zookeeper.connect=ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
另外2台相同配置，注意修改 broker.id、host.name参数即可
启动：
[hadoop@ruozedata003 config] sh kafka-server-start.sh -daemon ../config/server.properties
检查是否启动成功:
1.
[hadoop@ruozedata003 bin] tail -f server.log
[2021-04-12 21:03:42,084] INFO [GroupCoordinator 2]: Startup complete. (kafka.coordinator.GroupCoordinator)
[2021-04-12 21:03:42,101] INFO [Group Metadata Manager on Broker 2]: Removed 0 expired offsets in 8 milliseconds. (kafka.coordinator.GroupMetadataManager)
[2021-04-12 21:03:42,105] INFO [ThrottledRequestReaper-Produce], Starting (kafka.server.ClientQuotaManagerThrottledRequestReaper)
[2021-04-12 21:03:42,110] INFO Will not load MX4J, mx4j-tools.jar is not in the classpath (kafka.utils.Mx4jLoader$)
[2021-04-12 21:03:42,128] INFO Creating /brokers/ids/2 (is it secure? false) (kafka.utils.ZKCheckedEphemeral)
[2021-04-12 21:03:42,136] INFO Result of znode creation is: OK (kafka.utils.ZKCheckedEphemeral)
[2021-04-12 21:03:42,137] INFO Registered broker 2 at path /brokers/ids/2 with addresses: PLAINTEXT -> EndPoint(ruozedata003,9092,PLAINTEXT) (kafka.utils.ZkUtils)
[2021-04-12 21:03:42,138] WARN No meta.properties file under dir /home/hadoop/tmp/kafka-logs/meta.properties (kafka.server.BrokerMetadataCheckpoint)
[2021-04-12 21:03:42,160] INFO [Kafka Server 2], started (kafka.server.KafkaServer)
3.基础知识
producer 生产者 数据下发给kafka
broker 消息处理节点 kafka进程
consumer 消费者 从kafka消费数据
三个角色 是 对应三个组件
类似于 flume：
source
channel
sink
三个角色 是 对应一个组件-a agent 其实就是flume进程
topic：主题 相同属性的数据 放到一起
订单系统数据 omsdb --》maxwell--》kafka oms主题
仓储系统数据 wmsdb --》maxwell--》kafka wms主题
解析mysql binlog二进制文件数据 转成 json数据下发给kafka
手机银行 按事件划分
注册 --》register主题
登录 --》login主题
交易 --》deal主题
partition 分区 : 划分3个分区
是topic的物理的分组，一个主题必须至少有1个分区【有序】【分区内有序，但是全局无序 】
其实就是一个Linux文件夹而已，命名规则是topic名称-序号
ruozedata001: oms-0
ruozedata002: oms-1
ruozedata003: oms-2
HDFS文件 130M，划分2个块，128M 2M
replication 副本数 是和hdfs block的副本设计思想是一致的，为了高可靠 高容错
1台HDFS
只能1个副本 多个block块
4.常用命令
创建:
bin/kafka-topics.sh
--create
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
--replication-factor 3
--partitions 3
--topic g10
查看kafka集群多少个主题:
bin/kafka-topics.sh --list
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
查看指定主题的状况:
bin/kafka-topics.sh --describe
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
--topic g10
[hadoop@ruozedata001 kafka]$ bin/kafka-topics.sh --describe
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
--topic g10
Topic:g10 PartitionCount:3 ReplicationFactor:3 Configs:
Topic: g10 Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: g10 Partition: 1 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Topic: g10 Partition: 2 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
[hadoop@ruozedata001 kafka]$
Topic: g10 Partition: 0 Leader: 1 Replicas: 2,1,0 Isr: 1,0
Topic: g10 Partition: 1 Leader: 0 Replicas: 0,2,1 Isr: 0,1
Topic: g10 Partition: 2 Leader: 1 Replicas: 1,0,2 Isr: 1,0
Leader：是读写节点
Replicas: 2,4,8 是复制该分区数据的节点列表
ISR: 2,1,0 in-sync replicas 当前活跃的副本列表 是有可能成为leader
ack :
acks=0 If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won't generally know of any failures). The offset given back for each record will always be set to -1.
acks=1 This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost.
acks=all This means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee.
生产上 建议大家 就是使用all 先不管性能 先保障数据完全副本都成功写入
案例: 至关重要
bin/kafka-console-producer.sh
--broker-list ruozedata001:9092,ruozedata002:9092,ruozedata003:9092
--topic g10
bin/kafka-console-consumer.sh
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
--from-beginning
--topic g10
[hadoop@ruozedata002 kafka]$ bin/kafka-console-consumer.sh \
--zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka
--from-beginning
--topic g10
1
2
3
4
5
6
7
8
9
重新执行命令，模拟 【每个分区内有序，但是全局无序 】
[hadoop@ruozedata002 kafka]$ bin/kafka-console-consumer.sh --zookeeper ruozedata001:2181,ruozedata002:2181,ruozedata003:2181/kafka --from-beginning --topic g10
topic:0
3
6
9
topic:1
2
5
8
topic:2
1
4
7
业务背景: 实时数仓的 数据同步
上游mysql业务数据: ---》kafka
insert 100
update1 200
update2 150
update3 120
--delete
1--》0条
1--》1条
oms-0:
insert 100
update3 120
oms-1:
update1 200
--delete
oms-2:
update2 150
消费时就紊乱了 可能:
1--》1条 比业务多数
1--》1条 条数相同 但是值不同
补充: 每个分区内有序,为什么？
数据存储是磁盘文件 是追加模式 append
解决方案:
哪怕多个kafka节点，也只设置1个分区
所有的特征数据 id=1..100 都写到1个分区sql：
分组:
db,table,联合主键
排序:
时间戳-
【特征数据】： db,table,联合主键
去了解kafka写数据的代码，就能够再深入点
api: send(key,value)
send(null,insert....)
send(null,update1 ....)key:作用是什么，
就是将相同特征数据( db,table,联合主键)通过算法 写入到【同一个】分区，
【恰巧】分区内是有序的！
oms-0:
insert 100
update1 200
update2 150
update3 120
delete
算法:
hash("db1-t1-100") % 3= 102%3=34...0 下标 就是oms-0分区
1..100 平均写到三个分区
https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java
一个月 数据又对不上了 咋办？
max.in.flight.requests.per.connection 默认是5
1
2
3 失败了
4
5
3
乱序了
生产上 设置为 1
Allowing retries without setting max.in.flight.requests.per.connection to 1
will potentially change the ordering of records
retries 100次
MAXWELL:
PARTITION_BY: [ database | table | primary_key | transaction_id | column | random ]
写生产者：api
面试 其实也是面试官学习的过程 给他一个学习的机会 也就是你拿offer的机会
作业:
1.消息中间件除了kafka，还有哪些？大概的区别是什么 ？
2.http://kafka.apache.org/22/documentation.html#introduction
3.kafka集群部署 和 单机部署
4.maxwell 补充： https://www.bilibili.com/video/BV1zb411N7jj
5.ack参数 还有其他标识，是什么意思
6.because if two batches are sent to a single partition,
and the first fails and is retried but the second succeeds, then the records in the second batch may appear first.
面试多想想以自己公司为背景 不要偏的太离谱
