一. 安装和启动Kafka
我本地机器已经安装CDH 6.3.1版本，此处省略安装和启动Kafka的步骤。
Kafka版本:2.2.1
ps -ef|grep '/libs/kafka.\{2,40\}.jar'
1.1 Kafka的配置文件
[root@hp1 config]# find / -name server.properties
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/etc/kafka/conf.dist/server.properties
常用的配置如下:
#broker 的全局唯一编号，不能重复
broker.id=0
#删除 topic 功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的线程数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka 运行日志存放的路径
log.dirs=/opt/module/kafka/logs
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment 文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接 Zookeeper 集群地址
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181
二. 创建一个主题来存储事件
Kafka是一个分布式的事件流平台，可以让你跨多台机器读、写、存储和处理事件(在文档中也称为记录或消息)。
示例事件包括支付交易、来自移动电话的地理位置更新、发货订单、来自物联网设备或医疗设备的传感器测量，等等。这些事件被组织并存储在主题中。很简单，一个主题类似于文件系统中的一个文件夹，事件就是该文件夹中的文件。
2.1 创建主题
所以在你写你的第一个事件之前，你必须创建一个主题。打开另一个终端会话并运行:
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic first
2.2 查看当前事件描述
所有Kafka的命令行工具都有额外的选项:运行不带任何参数的Kafka -topics.sh命令来显示使用信息。例如，它还可以显示新主题的分区计数等详细信息:
-- 查看主题topic的描述
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic first
-- 查看所有的topic的描述
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181
一个分区一个副本
image.png
我们来看看创建多分区多副本
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic first_1_1
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic first_1_2
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 2 --topic first_2_2
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic first_3_3
image.png
本地测试只有3台broker，所以最多只能创建3个replication-factor
2.3 删除主题
需要 server.properties中设置 delete.topic.enable=true否则只是标记删除。 否则只是标记删除。
cd /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin
./kafka-topics.sh --zookeeper localhost:2181 --delete --topic first
三. 在主题中加入一些事件
Kafka客户端通过网络与Kafka的代理通信，用于写(或读)事件。一旦收到，代理将以持久和容错的方式存储事件，只要您需要—甚至永远。
运行控制台生成程序客户端，在主题中写入一些事件。默认情况下，您输入的每一行都将导致一个单独的事件被写入主题。
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-console-producer.sh --broker-list 10.31.1.124:9092 --topic first
image.png
四. 读事件
打开另一个终端会话并运行控制台消费者客户端来读取你刚刚创建的事件:
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/kafka-console-consumer.sh --from-beginning --bootstrap-server 10.31.1.124:9092 --topic first
image.png
--from-beginning：会把主题中以往所有的数据都读取出来。
您可以随时使用Ctrl-C停止客户端。
您可以自由地进行试验:例如，切换回您的生产者终端(上一步)来编写额外的事件，并查看这些事件如何立即显示在您的消费者终端上。
因为事件是持久性存储在Kafka中，它们可以被任意多的消费者读取。您可以通过再次打开另一个终端会话并再次运行前面的命令来轻松验证这一点。
六. 用kafka connect导入/导出你的数据作为事件流
您可能在现有系统(如关系数据库或传统消息传递系统)中有许多数据，以及许多已经使用这些系统的应用程序。Kafka Connect允许你不断地从外部系统获取数据到Kafka，反之亦然。因此，将现有系统与Kafka集成是非常容易的。为了使这个过程更容易，有数百个这样的连接器。
看看Kafka Connect部分，了解更多关于如何不断地导入/导出你的数据到Kafka。
七. 用kafka流处理你的事件
一旦你的数据以事件的形式存储在Kafka中，你就可以用Java/Scala的Kafka Streams客户端库来处理这些数据。它允许你实现关键任务实时应用和微服务，其中输入和/或输出数据存储在Kafka主题。Kafka Streams结合了客户端编写和部署标准Java和Scala应用程序的简单性和Kafka服务器端集群技术的优点，使这些应用程序具有高度的可扩展性、弹性、容错性和分布式。该库支持一次处理、有状态操作和聚合、窗口、连接、基于事件时间的处理等等。
网友评论