Kafka框架基础

Stella981
• 阅读 559

* Kafka框架基础

官网:kafka.apache.org

框架简介

Apache Kafka是分布式发布-订阅消息系统。它最初由LinkedIn公司开发,之后成为Apache项目的一部分。Kafka是一种快速、可扩展的、设计内在就是分布式的,分区的和可复制的提交日志服务。

相关概念

** 生产者

提供数据源生产的地方,对于同一个topic,生产者只能有一个,这样可以确保同一个topic数据来自同一个业务数据,支持多并发

** 消费者

消费数据的客户端,对于同一个topic,可以有多个消费者,比如spark,storm等等

** Broker

消息中间件处理结点,一个Kafka节点就是一个broker,多个broker可以组成一个Kafka集群。

** Topic

同一类消息的统称,Kafka集群能够同时负载多个topic分发。

** Partition

topic物理上的分组,一个topic可以分为多个partition,每个partition是一个有序的队列,同一个topic里面的数据是存放在不同的分区中。

** Replication

每个分区或者topic都是有副本的,副本的数量也是可以在创建topic的时候就指定好,保证数据的安全性,以及提供高并发读取效率。

** Segment

partition物理上由多个segment组成

** Offset

每个partition都由一系列有序的、不可变的消息组成,这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息

框架特色

** 同时为发布和订阅提供高吞吐量。Kafka每秒可以生产约25万消息(约50 MB),每秒处理55万消息(约110 MB)。

** 可进行持久化操作。将消息持久化到磁盘,因此可用于批量消费,例如ETL,以及实时应用程序。通过将数据持久化到硬盘以及replication防止数据丢失。

** 分布式系统,易于向外扩展。所有的producer、broker和consumer都会有多个,均为分布式的。无需停机即可扩展机器。

** 消息被处理的状态是在consumer端维护,而不是由server端维护。当失败时能自动平衡。

架构图

* 框架部署

** 相关下载

kafka以及scala:链接:http://pan.baidu.com/s/1pLBFJf1 密码:seto

** 解压Kafka以及scala

$ tar -zxf kafka_2.10-0.8.2.1.tgz -C /opt/modules/cdh/

$ tar -zxf scala-2.10.4.tgz -C /opt/modules/cdh/

** 安装JDK并配置环境变量

不再赘述

** 安装并启动zookeeper

在zookeeper的根目录下:

$ bin/zkServer.sh start

** 配置scala环境变量

# vi /etc/profile

$ source /etc/profile

(注意以上两条语句的执行用户)

添加如下:

##SCALA_HOME

SCALA_HOME=/opt/modules/cdh/scala-2.10.4

export PATH=$PATH:$SCALA_HOME/bin

使用命令检查scala配置是否正确:

$ scala -version,如图:

** 修改Kafka配置文件

server.properties

修改为如下:

producer.properties

变动内容如下:

consumer.properties

变动内容如下:

** 启动Kafka

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

** 创建Topic

$ bin/kafka-topics.sh --create --zookeeper z01:2181 --replication-factor 1 --partitions 1 --topic testTopic

** 启动生产者

$ bin/kafka-console-producer.sh --broker-list z01:9092 --topic testTopic

** 启动消费者

$ bin/kafka-console-consumer.sh --zookeeper z01:2181 --topic testTopic --from-beginning

在生产者窗口输入数据,在消费者窗口查看数据,测试如图:

消费者:

生产者:

* 整合测试

使用flume+kafka整合测试

** 配置flume

原来我们配置flume,是在tomcat所在机器节点开启了一个flume收集日志,并直接上传到HDFS,如果集群中存在多个机器节点,则势必导致对HDFS集群占用率过高,所以在面临多个flume集群时,一般会采用1~2个单独的flume节点来收集另外flume节点的日志,相当于弄了一个中转站,由中转站收集其他flume,再统一放置到HDFS系统中,此刻我们采用方案2,原理如图:

背景:在一台机器上开两个flume,分别收集tomcat日志和hive日志,这两者的日志信息分别输入到中间层flume(这个中间层flume也模拟在同一个机器节点上),然后中间层flume在将数据写入到HDFS。

首先检查一下hive的conf目录下的hive-log4j.properties配置中,是否已经指定了hive的日志目录,如果没有,请指定,如图:

涉及flume文件:以下文件存在于flume的conf目录下,如果不存在,请自行创建即可。

flume-apache-log-kafka.conf

flume-hive-log-kafka.conf

flume-connector-kafka.conf

依次启动:

a4:

$ bin/flume-ng agent --conf conf/ --name a4 --conf-file conf/flume-connector-kafka.conf

a3:

$ bin/flume-ng agent --conf conf/ --name a3 --conf-file conf/flume-hive-log-kafka.conf

a2:

$ bin/flume-ng agent --conf conf/ --name a2 --conf-file conf/flume-apache-log-kafka.conf

测试后如图,即可发现,日志在HDFS和kafka中都已经显示出来:


个人微博:http://weibo.com/seal13

QQ大数据技术交流群(广告勿入):476966007

作者:Z尽际链接:https://www.jianshu.com/p/a24af7a86392來源:简书著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。
点赞
收藏
评论区
推荐文章
blmius blmius
2年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
Jacquelyn38 Jacquelyn38
2年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
Stella981 Stella981
2年前
KVM调整cpu和内存
一.修改kvm虚拟机的配置1、virsheditcentos7找到“memory”和“vcpu”标签,将<namecentos7</name<uuid2220a6d1a36a4fbb8523e078b3dfe795</uuid
Easter79 Easter79
2年前
Twitter的分布式自增ID算法snowflake (Java版)
概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移
Wesley13 Wesley13
2年前
mysql设置时区
mysql设置时区mysql\_query("SETtime\_zone'8:00'")ordie('时区设置失败,请联系管理员!');中国在东8区所以加8方法二:selectcount(user\_id)asdevice,CONVERT\_TZ(FROM\_UNIXTIME(reg\_time),'08:00','0
Wesley13 Wesley13
2年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
2年前
Django中Admin中的一些参数配置
设置在列表中显示的字段,id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
Wesley13 Wesley13
2年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
3个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这