二级网站免费建,wordpress微信联系,网站建设到本地,网站正在升级建设中代码生产经验 面试重点 Broker面试重点 代码,开发重点 67 章了解
如何记录行为数据
1. Kafka概述
1.产生原因
前端 传到日志 日志传到Flume 传到HADOOP 但是如果数据特比大#xff0c;HADOOP就承受不住了
2.Kafka解决问题
控流消峰 Flume传给Kafka 存到Kafka Hadoop 从Kafka…生产经验 面试重点 Broker面试重点 代码,开发重点 67 章了解
如何记录行为数据
1. Kafka概述
1.产生原因
前端 传到日志 日志传到Flume 传到HADOOP 但是如果数据特比大HADOOP就承受不住了
2.Kafka解决问题
控流消峰 Flume传给Kafka 存到Kafka Hadoop 从Kafka取数据 而不是Kafka强行发 类似 菜鸟驿站 先存取来我们主动去取,或者指定他去送
存到HDFS的一定不是实时数据因为HDFS太慢了
3.应用场景
1. 缓冲/消峰
消息队列存储数据而不是直接发给处理系统处理完一部分再取再处理
2.解耦
通过中间件接口适配不同数据源和目的地 3.异步通信
允许用户将消息放入队列但不立即处理然后再需要的时候处理。 为什么异步处理快 同步需要等待 点餐 同步服务员过来给我点餐 ,这里需要服务员过来 异步扫桌子码自己点餐
4.消息队列模式 Kafka使用发布订阅模式 数据会保存一段时间
5.基础架构
生产者 - Broker - Group TopicA是什么 这里的分区是什么 分区 物理分割 为什么要分割结合集群分散存储 1. 分区操作
topic 是一整个数据 分区是为了将数据分在不同的Broker上。类似于HDFS Broker是物理存储 Partition 类似于DN
2.消费者组的概念
类似权限管理把组内并行消费便于管理
Producer生产者 向Kafka broker发消息的客户端(自主)Consumer消费者 从Kafka取消息的客户端自主Group 组 消费者组。消费者组内每个消费者负责消费不同分区的数据一个分区只能由一个组内消费者消费消费者组之间互不影响。所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者。 这个说明分区不支持并行读取一个分区只能同时一个组内消费者消费。 组内消费者对不同分区进行读取是为了优化读取速率.Broker 一台Kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。Topic可以理解为一个队列生产者和消费者面向的都是一个topic。Partition为了实现扩展性一个非常大的topic可以分布到多个broker即服务器上一个topic可以分为多个partition每个partition是一个有序的队列。Leader每个分区多个副本的“主”生产者发送数据的对象以及消费者消费数据的对象都是Leader。Follower每个分区多个副本中的“从”实时从Leader中同步数据保持和Leader数据的同步。Leader发生故障时某个Follower会成为新的Leader。Replica副本。一个topic的每个分区都有若干个副本一个Leader和若干个Follower。
3.副本
备份功能备份partition 多个副本 类似HDFS,类似DN(存入一下子3个)把 多副本时,一个副本叫leader 另一个副本叫follower, 也是选出来的角色 交互时只和leader交互 follower平时只有备份作用,但是当leader倒下时,他直接成为leader 这里是存储数据的目录,而不是存Kafka自己日志的目录 高可用 配置多个 replicas 是存储副本的位置 lsr 是目前存活的副本
分区数只能改大不能改小 副本数修改 通过JSON手动修改
消费者按最新的offect进行消费
5.配置
1.解压
[atguiguhadoop102 software]$ tar -zxvf kafka_2.12-3.3.1.tgz -C /opt/module/
[atguiguhadoop102 module]$ mv kafka_2.12-3.3.1/ kafka2.配置文件
配置文件目前只需要修改三个 broker编号 不同机器只需要编号不同即可 log.dir 数据存放位置 zookeeper.connect 连接集群的地址
[atguiguhadoop102 kafka]$ cd config/
[atguiguhadoop102 config]$ vim server.properties# 修改1 broker的全局唯一编号不能重复只能是数字。
broker.id0#处理网络请求的线程数量
num.network.threads3
#用来处理磁盘IO的线程数量
num.io.threads8
#发送套接字的缓冲区大小
socket.send.buffer.bytes102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes102400
#请求套接字的缓冲区大小
socket.request.max.bytes104857600
#修改2 这里其实是是存放到Kafka的数据的地方 kafka运行日志(数据)存放的路径路径不需要提前创建kafka自动帮你创建可以配置多个磁盘路径路径与路径之间可以用分隔
log.dirs/opt/module/kafka/datas
#topic在当前broker上的分区个数
num.partitions1
#用来恢复和清理data下数据的线程数量
num.recovery.threads.per.data.dir1
# 每个topic创建时的副本数默认时1个副本
offsets.topic.replication.factor1
#segment文件保留的最长时间超时将被删除
log.retention.hours168
#每个segment文件的大小默认最大1G
log.segment.bytes1073741824
# 检查过期数据的时间默认5分钟检查一次是否数据过期
log.retention.check.interval.ms300000
#修改3 连接集群的位置 配置连接Zookeeper集群地址在zk根目录下创建/kafka方便管理
zookeeper.connecthadoop102:2181,hadoop103:2181,hadoop104:2181/kafka3. 环境变量
sudo vim /etc/profile.d/my_env.sh#KAFKA_HOME
export KAFKA_HOME/opt/module/kafka
export PATH$PATH:$KAFKA_HOME/bin这个是为了启动的时候不需要输入一长串地址 比如bin/kafka-server-start.sh -daemon config/server.properties 这里的config是kafka的路径 启动需要输入全路径
bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties修改后
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties4.集群脚本
#! /bin/bash
if [ $# -lt 1 ]
then echo 参数错误请输入start或者stopexit
fi
case $1 in
start){for i in hadoop102 hadoop103 hadoop104 do echo ---------------启动 $i Kafka ----------------------ssh $i $KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.propertiesdone
};;
stop){for i in hadoop102 hadoop103 hadoop104do echo ---------------停止 $i Kafka ---------------------ssh $i $KAFKA_HOME/bin/kafka-server-stop.sh -daemon $KAFKA_HOME/config/server.propertiesdone
};;
esac2.命令
1.主题命令
1. --bootstrap-server String: server toconnect to
连接Broker 操作Kafka必须有这个命令 既可以输入一个也可以输入多个
kafka-topics.sh --bootstrap-server hadoop102:9092,hadoop103:9092,hadoop104:9092 2.主题的创建和删除
– create 空格 ±-topic空格主题名 – delete 空格 ±-topic空格主题名 主题 主题名 一般放最后
[atguiguhadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first–topic 定义topic名 –replication-factor 定义副本数 –partitions 定义分区数
3.查看所有主题
–list
4.查看主题详细描述
可以看单个主题可以看全部主题 不加后缀默认查看全部 查看单个需要空格 ±-topic空格主题名
5.修改–alter
设置分区数 –partitions Integer: # of partitions 分区只能调大不能调小
设置分区副本 –replication-factorInteger: replication factor
// 手动调整kafka topic分区的副本数 {// 1. 版本号 这个是自定义的版本号version:1,// 2. 分区是重点因为副本改变分区也要改变。// 其实就是将分区的副本重新进行布局partitions:[{topic:first,partition:0,replicas:[1,2,0]},{topic:first,partition:1,replicas:[2,0,1]},{topic:first,partition:2,replicas:[2,0,1]}]
}
// 运行命令
//kafka-reassign-partitions.sh --bootstrap-server hadoop102:9092 --reassignment-json-file ./rep.json --execute更新系统默认的配置。 –config String: namevalue 临时调配参数
2.生产者命令
1.操作
--topic String: topic
[atguiguhadoop102 kafka]$ bin/kafka-console-producer.sh
bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first3.消费者
kafka-console-consumer.sh --bootstrap-server 集群 主题
–bootstrap-server String: server toconnect to 连接的Kafka Broker主机名称和端口号。–topic String: topic 操作的topic名称。–from-beginning 从头开始消费。–group String: consumer group id 指定消费者组名称。
kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first为什么消费者未开启时,生产者发送的消息,等消费者起来了收不到? 没有指定消费者组时,每次开启,消费者属于的消费者组就是随机的,那么就无法进行断点续传 当主动指定组后,再次登录,在指定组后,会自动开启断点续传功能 想要提前的顺序,就需要,–from-beginning 但是不能和用户组一起跑
