Hadoop之Apache Kafka

Posted 程序猿的修身养性

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hadoop之Apache Kafka相关的知识,希望对你有一定的参考价值。

1、Kafka概述

Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编程语言编写,它是一个高吞吐量、支持分区(partition)的、多副本(replica)的、基于Zookeeper协调的、采用发布者-订阅者模型的分布式消息系统,最大的特点是可以实时处理大量数据以满足各种应用场景,诸如基于Hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎、Web/nginx日志、消息服务等等。

在大数据系统中,常常会碰到一个问题,整个大数据系统由多个子系统组成,数据需要在各个子系统之间高效率、低延迟地不停流转。传统的企业级消息系统并不是很适合大规模的数据处理,为了同时解决在线应用(消息)和离线应用(数据文件、日志)的需求,Kafka应运而生,它在这之中可以起到两个作用:

(1)降低各个系统之间组网的复杂度;

(2)降低编程复杂度,各个系统不再需要相互之间协商接口,而更像是插口插在插座上,Kafka承担了高速数据总线的作用。

2、Kafka的特性

Kafka作为一个分布式消息系统,它具有非常多优良的特性,其中几个最主要的列举如下:

  • 高吞吐量、低延迟:Kafka每秒可以处理几十万条消息,它的延迟最低只有几毫秒,每个Topic可以分成多个partition(分区),而consumer group(消费者组)对partition进行consume(消费)操作;

  • 可扩展性:Kafka集群支持热扩展;

  • 持久性、可靠性:消息被持久化到本地磁盘,并且支持数据备份防止数据丢失;

  • 容错性:允许集群中节点失败(若副本数量为n,则允许n-1个节点失败);

  • 高并发:支持数千个客户端同时读写。

3、Kafka的应用场景

Kafka是一个非常优秀的消息系统,它拥有的众多特性为其广泛应用打下了坚实的基础,主要应用领域列举如下:

  • 日志收集:日志收集方面,其实开源产品有很多,包括Scribe、Apache Flume等,很多人使用Kafka代替日志聚合(log aggregation)。日志聚合,一般来说是从服务器上收集日志文件,然后放到一个集中的位置(如文件服务器或HDFS)进行处理。然而,Kafka忽略掉文件的细节,将其更清晰地抽象成一个个日志或事件的消息流。这就让Kafka的处理过程延迟更低,更容易支持多数据源和分布式数据处理。比起以日志为中心的系统,如Scribe或者Flume来说,Kafka提供同样高效的性能和因为复制导致的更高的耐用性保证,以及更低的端到端延迟;

  • 消息队列:比起大多数的消息系统来说,Kafka有更好的吞吐量,内置的分区、冗余及容错性等,让Kafka成为一个很好的大规模消息处理的解决方案。消息系统一般吞吐量相对较低,但是需要更小的端到端延时,并常常依赖于Kafka提供的强大的持久性保障,在这个领域,Kafka足以媲美传统消息系统,如ActiveMR或RabbitMQ;

  • 行为跟踪:Kafka的另一个应用场景是跟踪用户的浏览页面、搜索及其他行为,然后以发布-订阅的模式实时记录到对应的Topic里。那么这些结果被订阅者拿到后,就可以做进一步的实时处理,或实时监控,或放到Hadoop/数据仓库中做离线分析和数据挖掘等;

  • 元信息监控:这通常是作为操作记录的监控模块来使用,即汇集记录一些操作信息,可以理解为运维性质的数据监控;

  • 流式处理:这个的应用场景非常多,也很好理解,保存采集的流数据,以提供之后对接的Storm或其他流式计算框架进行处理。很多用户会将那些从原始Topic来的数据进行阶段性处理、汇总、扩充或者以其他的方式转换到新的Topic下再继续后面的处理,比如一个文章推荐的处理流程,可能是先从RSS数据源中抓取文章的内容,然后将其丢入一个叫做“文章”的Topic中,后续操作可能是需要对这个内容进行清理,比如回复正常数据或者删除重复数据,最后再将内容匹配的结果返还给用户。这就在一个独立的Topic之外,产生了一系列的实时数据处理的流程,Storm和Samza是非常著名的实现这种类型数据转换的框架;

  • 事件源:事件源是一种应用程序设计的方式,该方式的状态转移被记录为按时间顺序排序的记录序列,Kafka可以存储大量的日志数据,这使得它成为一个对使用这种方式的应用来说绝佳的后台,比如动态汇总。

  • 持久性日志(commit log):Kafka可以为一种外部的持久性日志的分布式系统提供服务,这种日志可以在节点间备份数据,并为故障节点数据恢复提供一种重新同步的机制。Kafka中日志压缩功能为这种用法提供了条件,在这种用法中,Kafka类似于Apache Zookkeeper项目。

4、Kafka的基本架构

一个典型的Kafka体系结构包括若干Producer(可以是服务器日志、业务数据、网页前端产生的page view等等),若干Broker(Kafka支持水平扩展,一般Broker数量越多,集群吞吐率也越高),若干Consumer(支持消费者组Group),以及一个Zookeeper集群。这些部件组成的体系结构如下图所示:

Kafka的整体结构非常简单,是显示的分布式结构,Producer、Broker(Kafka主体)和Consumer都可以有多个。Producer和Consumer实现Kafka注册的接口,数据从Producer发送到Broker,Broker起到一个中间缓存和分发的作用,然后再由Broker分发给注册到系统中的Consumer。Broker的作用类似于缓存,即活跃的数据和离线处理系统之间的缓存,客户端和服务器端的通信,是基于简单、高性能且与编程语言无关的TCP协议。下面稍加解释这几个概念:

  • Topic:特指Kafka处理的消息源的不同分类,Kafka根据Topic对消息进行归类,发布到Kafka集群的每条消息都需要指定一个Topic;

  • Broker:消息中间件处理节点,也可以叫做缓存代理,一个Kafka节点就是一个Broker,一个或者多个Broker可以组成一个Kafka集群;

  • Producer:消息或数据的生产者,向Kafka的一个Topic发布消息的过程叫做Producer,可简单看作是向Broker发送消息的客户端;

  • Consumer:消息或数据的消费者,订阅Topics并处理发布的消息的过程叫做Consumer,可简单看作是从Broker读取消息的客户端;

  • Message:消息,通信的基本单位,每个Producer可以向一个Topic(主题)发布一些消息;

  • Partition:Topic物理上的分组,一个Topic可以分为多个Partition,每个Partition是一个有序的队列,Partition中的每条消息都会被分配一个有序的id(offset);

  • ConsumerGroup:每个Consumer属于一个特定的ConsumerGroup,一条消息可以发送到多个不同的ConsumerGroup,但是一个ConsumerGroup中只能有一个Consumer能够消息某个消息。

5、Kafka消息发送流程

Kafka中的消息发送流程同样非常简单,如下图所示:

Hadoop之Apache Kafka

  • Producer根据指定的Partition方法(如round-robin、hash等),将消息发布到指定的Topic的Partition里面;

  • Kafka集群接收到Producer发送过来的消息后,将其持久化到磁盘,并保留消息指定的时间(该数值可通过参数进行配置),而不关注消息是否被消费;

  • Consumer从Kafka集群pull数据,并控制获取消息的offset。

6、Kafka操作实践

总体来说,Kafka有三种安装模式,分别是单机单Broker方式、单机多Broker方式以及多机多Broker方式。下面分别进行介绍。

A、单机单Broker方式

将Kafka安装包kafka_2.9.2-0.8.1.1.tgz上传到主机hadoop221的/root/tools目录下,运行命令tar -zxvf kafka_2.9.2-0.8.1.1.tgz -C /root/training/,将Kafka安装包解压到/root/training/目录下,然后为Kafka配置环境变量,运行命令vi /root/.bash_profile进行编辑,在文件末尾添加如下内容:

KAFKA_HOME=/root/training/kafka_2.9.2-0.8.1.1

export KAFKA_HOME

 

PATH=$KAFKA_HOME/bin:$PATH

export PATH

保存退出,然后运行命令source /root/.bash_profile,使环境变量生效。接下来配置Kafka的核心配置文件server.properties,该文件位于config目录下,进入到/root/training/kafka_2.9.2-0.8.1.1/config目录下,运行命令vi server.properties编辑文件,将参数log.dirs的内容修改为:log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs,将参数zookeeper.connect修改为:zookeeper.connect=hadoop221:2181,保存退出;最后运行命令mkdir logs创建logs目录,用于存放Kafka运行过程中产生的log。

至此,Kafka的单机单Broker方式配置完成。下面进行实践操作,测试下Kafka的运行效果。首先,确保单机模式的Zookeeper已经启动起来,若还没有启动,则运行命令zkServer.shstart,启动Zookeeper;然后运行命令kafka-server-start.shconfig/server.properties &,以后台方式启动Kafka。输出的部分关键log,如下图所示:

Hadoop之Apache Kafka

创建Topic

运行命令:kafka-topics.sh --create --zookeeper hadoop221:2181-replication-factor 1 --partitions 3 --topic mytopic1,创建名称为mytopic1的Topic,输出的log如下图所示:

Hadoop之Apache Kafka

发送消息

运行命令:kafka-console-producer.sh --broker-list hadoop221:9092 --topic mytopic1,启动生产者线程。

接收消息

运行命令:kafka-console-consumer.sh --zookeeper hadoop221:2181--topic mytopic1,启动消费者线程。

在一个连接窗口中启动生产者线程,在另外两个连接窗口中启动分别启动消费者线程,如下图所示:

Hadoop之Apache Kafka

在生产者窗口中相继输入两行字符串,My best wishis that you love me(输完按回车键)和Hello World!(输完按回车键),在两个消费者窗口中相继显示出了这两条消息,如下图所示:

Hadoop之Apache Kafka

B、单机多Broker方式

前面已经安装和配置好了Kafka的单机单Broker方式,在其基础之上,现在来配置Kafka的单机多Broker方式。停止Kafka的运行,进入到/root/training/kafka_2.9.2-0.8.1.1/config目录下,依次运行命令:

cp server.properties server0.properties

cp server.properties server1.properties

cp server.properties server2.properties

cp server.properties server3.properties

保持配置文件server0.properties中的配置不变;配置文件server1.properties中相应修改如下:

broker.id=1

port=9093

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs1

并在目录/root/training/kafka_2.9.2-0.8.1.1下创建logs1目录;

配置文件server2.properties中相应修改如下:

broker.id=2

port=9094

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs2

并在目录/root/training/kafka_2.9.2-0.8.1.1下创建logs2目录;

配置文件server3.properties中相应修改如下:

broker.id=3

port=9095

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs3

并在目录/root/training/kafka_2.9.2-0.8.1.1下创建logs3目录;

最后,依次运行如下四条命令:

kafka-server-start.sh config/server0.properties &

kafka-server-start.sh config/server1.properties &

kafka-server-start.sh config/server2.properties &

kafka-server-start.sh config/server3.properties &

然后运行命令jps,如下图所示,可以看到启动了4个Kafka实例Broker。接下来创建主题,创建生产者线程及消费者线程完全同前面一致,这里不再赘述。

Hadoop之Apache Kafka

C、多机多Broker方式

这里先在主机hadoop222上进行Kafka的安装和配置,然后将配置好的Kafka目录通过网络拷贝到主机hadoop223和主机hadoop224上。

将Kafka安装包kafka_2.9.2-0.8.1.1.tgz上传到主机hadoop222的/root/tools目录下,运行命令tar -zxvfkafka_2.9.2-0.8.1.1.tgz -C /root/training/,将Kafka安装包解压到/root/training/目录下,然后配置Kafka的环境变量,运行命令vi /root/.bash_profile,在文件末尾添加如下内容:

KAFKA_HOME=/root/training/kafka_2.9.2-0.8.1.1

export KAFKA_HOME

 

PATH=$KAFKA_HOME/bin:$PATH

export PATH

保存退出,并运行命令source/root/.bash_profile,使环境变量生效。在主机hadoop223和主机hadoop224进行同样的操作,一并配置好Kafka的环境变量。

下面在主机hadoop222上配置Kafka的核心配置文件server.properties,进入到/root/training/kafka_2.9.2-0.8.1.1/config目录,依次运行命令:

cp server.properties server0.properties

cp server.properties server1.properties

cp server.properties server2.properties

配置文件server0.properties,修改如下:

broker.id=0

port=9092

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs0

zookeeper.connect=hadoop222:2181,hadoop223:2181,hadoop224:2181

配置文件server1.properties,修改如下:

broker.id=1

port=9093

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs1

zookeeper.connect= hadoop222:2181,hadoop223:2181,hadoop224:2181

配置文件server2.properties,修改如下:

broker.id=2

port=9094

log.dirs=/root/training/kafka_2.9.2-0.8.1.1/logs2

zookeeper.connect= hadoop222:2181,hadoop223:2181,hadoop224:2181

进入到/root/training/kafka_2.9.2-0.8.1.1目录,依次创建logs0,logs1,logs2三个目录。

最后依次运行如下两条命令,将配置好的Kafka分别拷贝到主机hadoop223和主机hadoop224的/root/training目录下:

scp -r/root/training/kafka_2.9.2-0.8.1.1/ root@hadoop223:/root/training

scp -r/root/training/kafka_2.9.2-0.8.1.1/ root@hadoop224:/root/training

注意,在同一台主机上,端口号port不能相同,因此依次配置为9092,9093,9094;同时,在不同主机上,broker.id均不能相同,因此在主机hadoop223上的三个配置文件server.properties中的broker.id依次配置为3,4,5,在主机hadoop224上的三个配置文件server.properties中的broker.id依次配置为6,7,8。

下面在三台主机上依次启动配置好的3个Kafka Broker实例,在主机hadoop222上,在主机hadoop223上,在主机hadoop224上依次运行命令:

kafka-server-start.sh config/server0.properties &

kafka-server-start.sh config/server1.properties &

kafka-server-start.sh config/server2.properties &

运行结束后,三台主机上运行进程的情况如下图所示:

Hadoop之Apache Kafka

在主机hadoop222上创建主题mytopic235,运行命令kafka-topics.sh --create--zookeeper hadoop222:2181 -replication-factor 1 --partitions 3 --topic topic1,界面输出log如下图所示,注意下面的两个窗口有时候会有log输出,有时候不输出,搞不懂,但不影响。

Hadoop之Apache Kafka

再在主机hadoop222上创建生产者线程,运行命令:kafka-console-producer.sh--broker-list hadoop222:9092 --topic topic1,在主机hadoop223和主机hadoop224上分别创建消费者线程,运行命令:kafka-console-consumer.sh--zookeeper hadoop222:2181 --topic topic1,输出log如下图所示:

最后在主机hadoop222窗口中输入字符串:Best wishes to you!,按回车键发送;接着再输入字符串:I missyou!,按回车键发送,界面运行效果如下图所示:



以上是关于Hadoop之Apache Kafka的主要内容,如果未能解决你的问题,请参考以下文章

Hadoop技术之Apache Hadoop集群搭建

HBase之集群状态

Hadoop之Apache Kafka

Hadoop技术之HDFS工作流程与机制Apache Hadoop概述

Ambari之 Hadoop 部署神器

Hbase之修改表结构