Kafka组件、安装、常用操作命令、分区计算及负载均衡等干货说明_卡夫卡组件

1.Kafka是什么:(集群)

• kafka集群不包含生产者和消费者.

(1)Kafka是一个开源消息系统,由Scala写成.

(2)Kafka是一个分布式消息队列:生产者,消费者的功能.

他提供了类似于JMS的特性,但是在设计实现上完全不同,此外它并不是JMS规范的实现.

(3)Kafka组件

• 对消息保存根据Topic进行归类.
• 消息发送者称为Producer,消息接收者称为Consumer.
• kafka集群有多个kafka实例组成,每个实例(server)称为broker.

• 注意:

无论是kafka集群,还是producer和consumer都依赖于zookeeper集群保存一些meta信息,用于保证系统可用性.

(4)Kafka的优点:

1)Kafka充当缓冲区,系统不会崩溃

以前,外部源系统的数据转换通常在晚上分批进行.
Kafka通过充当从源系统接收数据的中介,然后实时地将这些数据提供给目标系统来解决这个缓慢的多步骤过程.
更重要的是,系统不会因此而崩溃,因为Kafka是其独立的一组服务器(称为Kafka集群)

2)减少对多个集成的需求

从本质上讲,Kafka减少了对多个集成的需求,因为所有的数据都需要通过Kafka.可以认为成每一个生产系统和消耗系统创建一个与Apache Kafka的集成.

3)低延迟和高吞吐量(且读写速度快)

通过解耦数据流,Kafka允许在我们需要时使用数据,而在不需要缓慢集成的情况下,Kafka将延迟(或每个数据电价在所需的时间)减少到仅仅10ms(与其他集成相比减少约10倍或更多),这意味着我们可以快速,实时地提供数据.
Kafka还可以横向拓展到集群中的数百个代理(或服务器)来管理大数据.

• 案例:

有些公司每秒通过Kafka拥有数百万个数据点的高负载.例如:优步使用Kafka实时将汽车位置数据提供给他们的激增定价计算模型.

4)每个用户都可以访问数据

由于所有数据都集中在Kafka中,因此任何团队的数据访问都变得更加容易.

• 案例:

在过去，欺诈团队可能不得不与网络团队互动以获取特定类型的用户数据，因为它们在不同的目标系统上运行。
现在，欺诈团队将能够通过Kafka直接访问用户数据，以及其他提要，例如财务数据或网站交互

5)实时处理

Kafka可以处理实时数据通道.
由于我们需要找到一个技术部分来处理来自应用程序的实时消息,因此实时处理是选择Kafka的核心原因之一.

2.Kafka核心组件(重要)

kafka集群(文件存储系统,多副本)

Topic(可以理解为一个消息队列): 消息根据Topic进行归类.

• 一个kafka集群可以创建多个topic.

一个topic可以保存多种数据类型的数据,一种数据类型可以存放到一个topic中(但是不必须).

• 每个topic可以创建一个或多个分区.

分区的数量和副本的数量是在创建topic时指定的.
在后期,某个topic的分区数可以重新指定,只能由少变多,无法由多变少.

• 每个分区是由多个segment组成的

segment的大小是可以配置的.默认是1G.segment里有两种类型的文件(.index,.log).

index存放的是Log数据对应的索引(两者一一对应).
log文件存放的是数据.

• Kafka是有多副本机制的

原始数据和副本数据是不可以在同一个节点中的.

Broker:每个kafka实例(server).

• 一台Kafka服务器就是一个broker.
• 一个集群由多个broker组成.
• 一个broker可以容纳多个topic.

Partition:

• 为了实现拓展性,一个非常大的topic可以分布到多个broker(即服务器)上,一个topic可以分为多个partition,每个partition是一个有序的队列.
• partition中的每条消息都会被分配一个有序的id(offset).
• kafka只保证按一个partition中的顺序将消息发给consumer,不保证一个topic的整体(多个partition间)的顺序.

Offset

• Kafka的存储文件都是按照offset.kafka来命名的,用offset做名字的好处是方便查找.

• 如:
  你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka

zookeeper:依赖集群保存meta信息.

非kafka集群

Producer:发送消息者

• 生产者负责采集数据并把数据传输到Kafka的某个topic中,比如:

flume,java后台服务,shell脚本,python脚本,logstash.

• 生产者是由多个进程组成(可以由多个生产者组成),一个生产者可以作为一个独立的进程,可以独立的分发数据.
• 多个生产者发送的数据是可以存储到同一个topic的同一个partition中.
• 一个生产者的数据也可以放到多个topic中.

Consumer:消息接受者(Consumer Group)

• 消费者负责消费(拉取)数据

flume从kafka中拉取数据存放到hdfs中,SparkStreaming,Storm…

• 一个Consumer Group也被称为Consumer集群(没有集群概念,习惯叫法)
• 新增或减少Consumer的数量会触发Kafka的负载均衡.
• ConsumerGroup可以消费一个或多个分区的数据,相反一个分区的数据只能被一个Consumer消费.如果想要消费,必须进行线程等待.
• ConsumerGroup成员之间消费的数据各不相同,在同一个group中,数据是不可以重复消费(一次仅一次的语义).

3.Kafka的安装

1.下载安装包

http://kafka.apache.org/downloads.html

2.将安装包放入到指定的目录下

3.解压安装包到指定目录下

tar -zxvf /data/kafka_2.11-0.9.1.tgz -C /data/
1

4.将kafka目录重命名

mv /data/kafka_2.11-0.9.1 /data/kafka
1

5.进入config目录进行修改配置文件

分别修改consumer.properties,producer.properties,server.properties.
这三个文件的修改在文件夹中查看

6.进行环境变量配置

vi /etc/profile
//先将Kafka目录导入
export KAFKA_HOME=/data/kafka
//ATH后加入
:$KAFKA_HOME/bin
1
2
3
4
5

4.Kafka操作命令

kafka启动命令:

先启动zookeeper

zkServer.sh start
1

依次在各节点上启动kafka

nohup kafka-server-start.sh data/kafka/config/server.properties &
1

kafka操作命令:

查看当前服务器中所有的topic

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

创建topic

//创建了一个zhangyonggang的topic
kafka-topics.sh --create --zookeeper hadoop1:2181 --replication-factor 1 --partitions 1 --topic loadkafkadata
1
2

replication-factor

当某个topic的replication-factor为N且N大于1时，每个Partition都会有N个副本(Replica)。kafka的replica包含leader与follower

删除topic

• 注意:

需要在server.properties中设置delete.topic.enable=true才是真的彻底删除,否则只是删除标记或者直接重启.

//删除一个zhangyonggang的topic
kafka-topics.sh --delete --zookeeper hadoop1:2181 --topic zhangyonggang
1
2

通过shell命令发送信息

• 经过测试:
• 在hadoop1开启kafka-console-producer.sh时,当hadoop1改为hadoop2或者hadoop3时,hadoop2或者hadoop3只要开启了kafka-console-consumer.sh,都可以接收到producer传来的消息.

kafka-console-producer.sh --broker-list hadoop1:9092 --topic hello
1

通过shell消费信息

• 经过测试:
• 在hadoop2和hadoop3同时开启kafka-console-consumer.sh,也可以同时接收producer传来的消息.

kafka-console-consumer.sh --zookeeper hadoop3:2181 --from-beginning --topic hello
1

可以认为:

在同一个zk上,开启producer一方为生产者,开启consumer一方或多方为消费者,都可以接收生产者发来的消息.

查看某个Topic的详情

kafka-topics.sh --topic test --describe --zookeeper hadoop1:2181
//结果如下:
1
2

对分区数进行修改

//将liuming这个topic的分区修改为4个
kafka-topics.sh --zookeeper hadoop1 --alter --partition 4 --topic liuming
1
2

---

Logs日志查看

先进入到自定义的Logs目录下,具体操作如下图:

• index中存放的是索引
• logs存放的是数据

多分区不能往少个分区分,会报错.
只能少分区向多分区中分.

5.Kafka的几个问题:

1. segment

• 每个topic的分区中有多个segment
• 一个分区会被分成相同大小数据数量不等的segment,数据的生命周期就是指的segment的生命周期.

2. 数据的存储机制:

• 首先是Broker接收到数据,将数据放到操作系统的缓存里(pagecache).
• pagecache会尽可能多的使用空闲内存,使用sendfile技术尽可能多的减少操作系统和应用程序之间的重复缓存.
• 写数据的时候是顺序写入,顺序写入的速度可达600M/S.

3.kafka是怎么解决负载均衡的?

1. 首先获取Consumer消费的起始分区号.
2. 然后计算出Consumer要消费的分区数量.
3. 用起始分区号的hash值%分区数.

分区数的计算

• 分区的数量最好是consumer的数量的整数倍(2-3倍最佳)
• consumer的数量
  一个线程对应一个consumer
• topic中有多个分区的情况下,数据是无法全局有序的进行存储的.但是可以做到分区内有序,因为存储时是按照数据一一存储的,数据有序即可分区内存储有序.
• 如多想要做到有序,可以试着将多个分区变为一个,但是牺牲了吞吐量.

4.数据是怎么分区的?

• kafka默认调用自己的分区器(DefaultPartitioner),当然也可以自定义分区器,需要实现Partition特质,实现Partition方法.

5.怎么保证数据不丢失?

• kafka的多副本机制就保证了数据的安全性,保证了数据的不丢失,副本数是在创建topic时指定的.

Spark的Streaming的Receiver方式和Direct方式的区别(重点)

Kafka的存储和实时处理(蓝色框内)