四.Spark Streaming之kafka单节点单broker部署及使用(一)

墨蓝 2022-04-16 04:12 188阅读 0赞

**一.介绍**

1).概念:  
    kafka是用于构建实时数据管道和流媒体应用。它是水平扩展的，容错的，快速的，并且在成千上万的公司中使用。消息中间件: 生产者和消费者

2).举例说明:  
妈妈: 生产者  
你 : 消费者  
馒头: 数据流

正常情况下: 生产一个 消费一个  
其他情况:  
     一直生产,你吃到某一个馒头时,你卡住(机器故障),馒头就丢失了  
     一直生产,做馒头速度快,你吃来不及,馒头也就丢失了

拿个碗/篮子,馒头做好以后先放在篮子里,你要吃的时候去篮子里面取出来吃

篮子/筐(kafka服务):  
当篮子满了,馒头就装不下了,咋办?  
多准备几个篮子,相当于就是kafka的扩容

3). 一个流处理平台应该最基本的具备以下三个能力:

1.能够让你订阅和发布你的流的记录,类似于消息队列
         2.以容错的方式存储记录
         3.当这个记录产生的时候,能让你时时刻刻处理这个记录

4).kafka的优势:

1.构建时时的流数据的管道,从系统和应用程序之间获取数据.
     2.构建时时的流处理程序能够转换和产生一些结果

5).kafka架构及核心概念:

producer: 生产者,就是生产馒头的(老妈)
    
         consumer: 消费者,就是吃馒头的(你)
    
         broker: 篮子(多个server)
    
         topic:  主题,给馒头打一个标签,比如:topic a的馒头是给你吃的,topic b的馒头时给你弟弟吃的(打标签进行区分)
    
         核心概念:
             1) kafka可以以一个或者多个server运行在集群上面
             2) 给数据记录打标签,消费者可以选择性的进行消费
             3) 每一条记录都有一个key,value和一个时间戳组成.

**二.zookeeper的安装配置**  
1).在搭建kafka之前首先要搭建zookeeper的环境,搭建过程如下:

wget  http://archive.cloudera.com/cdh5/cdh/5/zookeeper-3.4.5-cdh5.7.0.tar.gz

下载完成之后,通过tar命令解压缩,配置全局环境变量,然后开始对配置文件进行修改,找到conf下的zoo\_sample.cfg复制一份取名zoo.cfg,打开其进行配置文件的修改:

dataDir=/root/data/tmp/zk

创建一个已经存在文件夹路径,作为zookeeper数据文件的存储路径即可。  
2).启动zookeeper服务端和客户端进入到bin目录下:  
启动服务端:

./zkServer.sh start

启动客户端:

./zkCli.sh

3).简单操作:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]

**三.kafka的环境搭建及使用**  
1)首先是下载二进制文件,一定要注意,下载的时候要对应自己电脑上的scala的版本,因为kafka是scala编写的,如果版本不统一,后续会出问题:

wget https://archive.apache.org/dist/kafka/0.9.0.0/kafka_2.10-0.9.0.0.tgz

解压缩之后配置全局环境变量,然后开始修改配置文件,找到安装文件中的config目录下的server,properties文件,打开进行如下修改:

把host.name的注释解掉,改成本机的ip
       host.name=hadoop000
       日志存储路径
       log.dirs=/root/data/tmp/kafka-logs
       分区数量:
       num.partitions=1 暂不做修改
       zookeeper地址:
       zookeeper.connect=hadoop000:2181

完成以上配置之后,就可以启动kafka了。  
2).启动命令:

kafka-server-start.sh $KAFKA_HOME/config/server.properties

启动成功的标志:  
![在这里插入图片描述][2018111615465126.png]  
或者可以通过jps -m 查看是否有该进程:  
![在这里插入图片描述][20181116154834826.png]

3).创建topic(指定zookeeper):

kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic hello_topic

–replication-factor     表示副本系数  
–partitions                表示分区数量  
–topic                       表示标签(topic)的名字  
![在这里插入图片描述][20181116155825772.png]  
查看zookeeper中的创建的标签:

kafka-topics.sh --list --zookeeper hadoop000:2181

![在这里插入图片描述][20181116160004962.png]

4).发送消息(生产者指定broker):

kafka-console-producer.sh --broker-list hadoop000:9092 --topic hello_topic

启动之后的界面如下:  
![在这里插入图片描述][20181116161721978.png]  
5).开始消费(消费者指定zookeeper):

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic --from-beginning

–from-beginning 表示从头消费,就算你断开了,等你再次消费的时候,以前的消费记录还在,如果不加就表示从你当前消费开始记录。  
![在这里插入图片描述][20181116162052398.png]

注意:  
官网给的不准确,消费的应该是zookeeper中存在的去消费。

6).测试实时生产和消费:

生产者生产几条数据:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]  
查看消费者是否消费:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 2]

7).扩展:  
查看所有topic的详细信息:

kafka-topics.sh --describe --zookeeper hadoop000:2181

显示效果:  
![在这里插入图片描述][20181116163136941.png]

以上就完成了kafka单节点单broker的搭建和基本的使用,虽然有点麻烦,但是完成之后感觉就是一些基本的配置,详细的资料参考[  
官网][Link 1],里面有的命令不太准确,善用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70]: /images/20220416/5b0172ee20464f6997e19f14fd99ecd3.png
[2018111615465126.png]: /images/20220416/95fe5b83a07f4409b0e3c211745649e7.png
[20181116154834826.png]: /images/20220416/a60798f41d9e4f7e814cb148a7c01edb.png
[20181116155825772.png]: /images/20220416/0a8c7c2f85694217986f9a7741a01db6.png
[20181116160004962.png]: /images/20220416/184df8c8d7324fa9883826bb06ef234f.png
[20181116161721978.png]: /images/20220416/7296fbe7945940abab7a8d3d96074912.png
[20181116162052398.png]: /images/20220416/f927bd174c574076a5aeb1e8378e01a7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 1]: /images/20220416/f4e77c16a82e49308b4e3b628ffa00df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZ2hfc3Fo_size_16_color_FFFFFF_t_70 2]: /images/20220416/a1d5aa8d47e34911a722204b3314460e.png
[20181116163136941.png]: /images/20220416/fc6a7e7755a7462b88a5c591ef40482b.png
[Link 1]: http://kafka.apache.org/quickstart