第89讲:Spark Streaming on Kafka解析和安装实战

本课分2部分讲解:

第一部分,讲解Kafka的概念、架构和用例场景;

第二部分,讲解Kafka的安装和实战。

由于时间关系,今天的课程只讲到如何用官网的例子验证Kafka的安装是否成功。后续课程会接着讲解如何集成Spark Streaming和Kafka。

一、Kafka的概念、架构和用例场景

http://kafka.apache.org/documentation.html#introdution

1、Kafka的概念

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

什么是消息组件:

以帅哥和美女聊天为例,帅哥如何和美女交流呢?这中间通常想到的是微信、QQ、电话、邮件等通信媒介,这些通信媒介就是消息组件,帅哥把聊天信息发送给消息组件、消息组件将消息推送给美女,这就是常说的生产者、消费者模型。而且在发送信息时可以将内容进行分类,即所谓的Topic主题。Kafka就是这样的通信组件,将不同对象组件粘合起来的纽带,且是解耦合方式传递数据。

Apache Kafka与传统消息系统相比,有以下不同的特点:

  • 分布式系统,易于向外扩展;
  • 在线低延迟,同时为发布和订阅提供高吞吐量;
  • 将消息存储到磁盘,因此可以处理1天甚至1周前内容

2、Kafka的架构

Kafka既然具备消息系统的基本功能,那么就必然会有组成消息系统的组件:

Topic,Producer和Consumer。Kafka还有其特殊的Kafka Cluster组件。

Topic主题:

代表一种数据的类别或类型,工作、娱乐、生活有不同的Topic,生产者需要说明把说明数据分别放在那些Topic中,里面就是一个个小对象,并将数据数据推到Kafka,消费者获取数据是pull的过程。一组相同类型的消息数据流。这些消息在Kafka会被分区存放,并且有多个副本,以防数据丢失。每个分区的消息是顺序写入的,并且不可改写。

-       Producer(生产者):把数据推到Kafka系统的任何对象。

- Kafka Cluster(Kafka集群):把推到Kafka系统的消息保存起来的一组服务器,也叫Broker。因为Kafka集群用到了Zookeeper作为底层支持框架,所以由一个选出的服务器作为Leader来处理所有消息的读和写的请求,其他服务器作为Follower接受Leader的广播同步备份数据,以备灾难恢复时用。

- Consumer(消费者):从Kafka系统订阅消息的任何对象。

消费者可以有多个,并且某些消费者还可以组成Consumer Group。多个Consumer Group之间组成消息广播的关系,所以各个Group可以拉相同的消息数据。在Consumer Group内部,各消费者之间对Consumer Group拉出来的消息数据是队列先进先出的关系,某个消息数据只能给该Group的一个消费者使用。

数据传输基于kernel(内核)级别的(传输速度接近0拷贝-ZeroCopy)、没有用户空间的参与。Linux本身是软件,软件启动时第一个启动进程叫init,在init进程启动后会进入用户空间;例如:在分布式系统中,机器A上的应用程序需要读取机器B上的Java服务数据,由于Java程序对应的JVM是用户空间级别而且数据在磁盘上,A上应用程序读取数据时会首先进入机器B上的内核空间再进入机器B的用户空间,读取用户空间的数据后,数据再经过B机器上的内核空间分发到网络中,机器A网卡接收到传输过来的数据后再将数据写入A机器的内核空间,从而最终将数据传输给A的用户空间进行处理。如下图:

外部系统从Java程序中读取数据,传输给内核空间并依赖网卡将数据写入到网络中,从而把数据传输出去。其实Java本身是内核的一层外衣,Java Socket编程,操作的各种数据都是在JVM的用户空间中进行的。而Kafka操作数据是放在内核空间的,通常内核空间处理数据的速度比用户空间快上万倍,所以通过kafka可以实现高速读、写数据。

3、Kafka的用例场景

类似微信,手机和邮箱等等这样大家熟悉的消息组件,Kafka也可以:

-       支持文字/图片

-       可以存储内容

-       分门别类

从内容消费的角度,Kafka把邮箱中的邮件看成是Topic。

二、Kafka的安装和实战

http://kafka.apache.org/documentation.html#quickstart

1、安装和配置Zookeeper

Kafka集群模式需要提前安装好Zookeeper。

-       提示:Kafka单例模式不需要安装额外的Zookeeper,可以使用内置的Zookeeper。

-       Kafka集群模式需要至少3台服务器。本课实战用到的服务器Hostname:master,slave1,slave2。

-       本课中用到的Zookeeper版本是Zookeeper-3.4.6。

1)    下载Zookeeper

进入http://www.apache.org/dyn/closer.cgi/zookeeper/,你可以选择其他镜像网址去下载,用官网推荐的镜像:http://mirror.bit.edu.cn/apache/zookeeper/。提示:可以直接下载群里的Zookeeper安装文件。

下载zookeeper-3.4.6.tar.gz

1)    安装Zookeeper

提示:下面的步骤发生在master服务器。

以ubuntu14.04举例,把下载好的文件放到/root目录,用下面的命令解压:

cd /root

tar -zxvf zookeeper-3.4.6.tar.gz

解压后在/root目录会多出一个zookeeper-3.4.6的新目录,用下面的命令把它剪切到指定目录即安装好Zookeeper了:

cd /root

mv zookeeper-3.4.6 /usr/local/spark

之后在/usr/local/spark目录会多出一个zookeeper-3.4.6的新目录。下面我们讲如何配置安装好的Zookeeper。

2)    配置Zookeeper

提示:下面的步骤发生在master服务器。

  1. 配置.bashrc

-       打开文件:vi /root/.bashrc

-       在PATH配置行前添加:

export ZOOKEEPER_HOME=/usr/local/spark/zookeeper-3.4.6

-       最后修改PATH:

export PATH=${JAVA_HOME}/bin:${ZOOKEEPER_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${SCALA_HOME}/bin:${SPARK_HOME}/bin:${SPARK_HOME}/sbin:${HIVE_HOME}/bin:${KAFKA_HOME}/bin:$PATH

-       使配置的环境变量立即生效:source /root/.bashrc

  1. 创建data目录

-       cd $ZOOKEEPER_HOME

-       mkdir data

  1. 创建并打开zoo.cfg文件

-       cd $ZOOKEEPER_HOME/conf

-       cp zoo_sample.cfg zoo.cfg

-       vi zoo.cfg

  1. 配置zoo.cfg

# 配置Zookeeper的日志和服务器身份证号等数据存放的目录。

# 千万不要用默认的/tmp/zookeeper目录,因为/tmp目录的数据容易被意外删除。

dataDir=../data

# Zookeeper与客户端连接的端口

clientPort=2181

# 在文件最后新增3行配置每个服务器的2个重要端口:Leader端口和选举端口

# server.A=B:C:D:其中 A 是一个数字,表示这个是第几号服务器;

# B 是这个服务器的hostname或ip地址;

# C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口;

# D 表示的是万一集群中的 Leader 服务器挂了,需要一个端口来重新进行选举,

# 选出一个新的 Leader,而这个端口就是用来执行选举时服务器相互通信的端口。

# 如果是伪集群的配置方式,由于 B 都是一样,所以不同的 Zookeeper 实例通信

# 端口号不能一样,所以要给它们分配不同的端口号。

server.1=master:2888:3888

server.2=slave1:2888:3888

server.3=slave2:2888:3888

  1. 创建并打开myid文件

-       cd $ZOOKEEPER_HOME/data

-       touch myid

-       vi myid

  1. 配置myid

按照zoo.cfg的配置,myid的内容就是1。

3)    同步master的安装和配置到slave1和slave2

-       在master服务器上运行下面的命令

cd /root

scp ./.bashrc [email protected]:/root

scp ./.bashrc [email protected]:/root

cd /usr/local/spark

scp -r ./zookeeper-3.4.6 [email protected]:/usr/local/spark

scp -r ./zookeeper-3.4.6 [email protected]:/usr/local/spark

-       在slave1服务器上运行下面的命令

vi $ZOOKEEPER_HOME/data/myid

按照zoo.cfg的配置,myid的内容就是2。

-       在slave2服务器上运行下面的命令

vi $ZOOKEEPER_HOME/data/myid

按照zoo.cfg的配置,myid的内容就是3。

4)    启动Zookeeper服务

-       在master服务器上运行下面的命令

zkServer.sh start

-       在slave1服务器上运行下面的命令

source /root/.bashrc

zkServer.sh start

-       在slave1服务器上运行下面的命令

source /root/.bashrc

zkServer.sh start

5)    验证Zookeeper是否安装和启动成功

-       在master服务器上运行命令:jps和zkServer.sh status

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# jps

3844 QuorumPeerMain

4790 Jps

zkServer.sh status

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# zkServer.sh status

JMX enabled by default

Using config: /usr/local/spark/zookeeper-3.4.6/bin/../conf/zoo.cfg

Mode: follower

-       在slave1服务器上运行命令:jps和zkServer.sh status

source /root/.bashrc

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# jps

3462 QuorumPeerMain

4313 Jps

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# zkServer.sh status

JMX enabled by default

Using config: /usr/local/spark/zookeeper-3.4.6/bin/../conf/zoo.cfg

Mode: follower

-       在slave2服务器上运行命令:jps和zkServer.sh status

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# jps

4073 Jps

3277 QuorumPeerMain

[email protected]:/usr/local/spark/zookeeper-3.4.6/bin# zkServer.sh status

JMX enabled by default

Using config: /usr/local/spark/zookeeper-3.4.6/bin/../conf/zoo.cfg

Mode: leader

至此,代表Zookeeper已经安装和配置成功。

2、安装和配置Kafka

本课中用到的Kafka版本是Kafka-2.10-0.9.0.1。

1)    下载Kafka

进入http://kafka.apache.org/downloads.html,左键单击kafka_2.10-0.9.0.1.tgz。提示:可以直接下载群里的Kafka安装文件。

下载kafka_2.10-0.9.0.1.tgz

1)    安装Kafka

提示:下面的步骤发生在master服务器。

以ubuntu14.04举例,把下载好的文件放到/root目录,用下面的命令解压:

cd /root

tar -zxvf kafka_2.10-0.9.0.1.tgz

解压后在/root目录会多出一个kafka_2.10-0.9.0.1的新目录,用下面的命令把它剪切到指定目录即安装好Kafka了:

cd /root

mv kafka_2.10-0.9.0.1 /usr/local

之后在/usr/local目录会多出一个kafka_2.10-0.9.0.1的新目录。下面我们讲如何配置安装好的Kafka。

2)    配置Kafka

提示:下面的步骤发生在master服务器。

  1. 配置.bashrc

-       打开文件:vi /root/.bashrc

-       在PATH配置行前添加:

export KAFKA_HOME=/usr/local/kafka_2.10-0.9.0.1

-       最后修改PATH:

export PATH=${JAVA_HOME}/bin:${ZOOKEEPER_HOME}/bin:${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:${SCALA_HOME}/bin:${SPARK_HOME}/bin:${SPARK_HOME}/sbin:${HIVE_HOME}/bin:${KAFKA_HOME}/bin:$PATH

-       使配置的环境变量立即生效:source /root/.bashrc

  1. 打开server.properties

-       cd $ZOOKEEPER_HOME/config

-       vi server.properties

  1. 配置server.properties

broker.id=0

port=9092

zookeeper.connect=master:2181,slave1:2181,slave2:2181

3)    同步master的安装和配置到slave1和slave2

-       在master服务器上运行下面的命令

cd /root

scp ./.bashrc [email protected]:/root

scp ./.bashrc [email protected]:/root

cd /usr/local

scp -r ./kafka_2.10-0.9.0.1 [email protected]:/usr/local

scp -r ./kafka_2.10-0.9.0.1 [email protected]:/usr/local

-       在slave1服务器上运行下面的命令

vi $KAFKA_HOME/config/server.properties

修改broker.id=1。

-       在slave2服务器上运行下面的命令

vi $KAFKA_HOME/config/server.properties

修改broker.id=2。

4)    启动Kafka服务

-       在master服务器上运行下面的命令

cd $KAFKA_HOME/bin

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

-       在slave1服务器上运行下面的命令

source /root/.bashrc

cd $KAFKA_HOME/bin

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

-       在slave2服务器上运行下面的命令

source /root/.bashrc

cd $KAFKA_HOME/bin

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

5)    验证Kafka是否安装和启动成功

-       在任意服务器上运行命令创建Topic“HelloKafka”:

kafka-topics.sh --create --zookeeper master:2181,slave1:2181,slave2:2181 --replication-factor 3 --partitions 1 --topic HelloKafka

-       在任意服务器上运行命令为创建的Topic“HelloKafka”生产一些消息:

kafka-console-producer.sh --broker-list master:9092,slave1:9092,slave2:9092 --topic HelloKafka

输入下面的消息内容:

This is DT_Spark!

I’m Rocky!

Life is short, you need Spark!

-       在任意服务器上运行命令从指定的Topic“HelloKafka”上消费(拉取)消息:

kafka-console-consumer.sh --zookeeper master:2181,slave1:2181,slave2:2181 --from-beginning --topic HelloKafka

过一会儿,你会看到打印的消息内容:

This is DT_Spark!

I’m Rocky!

Life is short, you need Spark!

-       在任意服务器上运行命令查看所有的Topic名字:

kafka-topics.sh --list --zookeeper master:2181,slave1:2181,slave2:2181

-       在任意服务器上运行命令查看指定Topic的概况:

kafka-topics.sh --describe --zookeepermaster:2181,slave1:2181,slave2:2181 --topic HelloKafka

至此,代表Kafka已经安装和配置成功。

总结:

使用Spark Streaming可以处理各种数据来源类型,如:数据库、HDFS,服务器log日志、网络流,其强大超越了你想象不到的场景,只是很多时候大家不会用,其真正原因是对Spark、spark streaming本身不了解。

时间: 2024-10-29 04:25:36

第89讲:Spark Streaming on Kafka解析和安装实战的相关文章

第89课:Spark Streaming on Kafka解析和安装实战

本课分2部分讲解: 第一部分,讲解Kafka的概念.架构和用例场景: 第二部分,讲解Kafka的安装和实战. 由于时间关系,今天的课程只讲到如何用官网的例子验证Kafka的安装是否成功.后续课程会接着讲解如何集成Spark Streaming和Kafka. 一.Kafka的概念.架构和用例场景 http://kafka.apache.org/documentation.html#introdution 1.Kafka的概念 Apache Kafka是分布式发布-订阅消息系统.它最初由Linked

第89课:SparkStreaming on Kafka之Kafka解析和安装实战

本篇博文将从以下方面组织内容: 1. Kafka解析 2. 消息组件Kafka 3. Kafka安装 实验搭建所需要的软件: kafka_2.10-0.9.0.1 Zookeeper集群已经安装好.在上一篇博文有安装步骤,不清楚的朋友可以参考下. 一:Kafka解析 1. Kafka是生产者和消费者模式中广播概念,Kafka也可以实现队列的方式. 2. Kafka不仅是一个消息中间键,还是一个存储系统,可以将流进来的数据存储一段时间.这就与传统的流式处理不一样,传统的流式处理处理完数据之后就消失

SBT 构建 spark streaming集成kafka (scala版本)

前言: 最近在研究spark 还有 kafka , 想通过kafka端获取的数据,利用spark streaming进行一些计算,但搭建整个环境着实不易,故特此写下该过程,分享给大家,希望大家可以少走点弯路,能帮到大家! 环境准备:    操作系统 : ubuntu14.04 LTS hadoop 2.7.1   伪分布式搭建 sbt-0.13.9 kafka_2.11-0.8.2.2 spark-1.3.1-bin-hadoop2.6 scala 版本 : 2.10.4 注: 请重视版本问题,

Spark Streaming使用Kafka保证数据零丢失

来自: https://community.qingcloud.com/topic/344/spark-streaming使用kafka保证数据零丢失 spark streaming从1.2开始提供了数据的零丢失,想享受这个特性,需要满足如下条件: 数据输入需要可靠的sources和可靠的receivers 应用metadata必须通过应用driver checkpoint WAL(write ahead log) 可靠的sources和receivers spark streaming可以通过

Spark Streaming和Kafka整合保证数据零丢失

当我们正确地部署好Spark Streaming,我们就可以使用Spark Streaming提供的零数据丢失机制.为了体验这个关键的特性,你需要满足以下几个先决条件: 1.输入的数据来自可靠的数据源和可靠的接收器: 2.应用程序的metadata被application的driver持久化了(checkpointed ); 3.启用了WAL特性(Write ahead log). 下面我将简单地介绍这些先决条件. 可靠的数据源和可靠的接收器 对于一些输入数据源(比如Kafka),Spark S

spark streaming 对接kafka记录

spark streaming 对接kafka 有两种方式: 参考: http://group.jobbole.com/15559/ http://blog.csdn.net/kwu_ganymede/article/details/50314901 Approach 1: Receiver-based Approach 基于receiver的方案: 这种方式使用Receiver来获取数据.Receiver是使用Kafka的高层次Consumer API来实现的.receiver从Kafka中获

Spark Streaming、Kafka结合Spark JDBC External DataSouces处理案例

场景:使用Spark Streaming接收Kafka发送过来的数据与关系型数据库中的表进行相关的查询操作: Kafka发送过来的数据格式为:id.name.cityId,分隔符为tab 1 zhangsan 1 2 lisi 1 3 wangwu 2 4 zhaoliu 3 MySQL的表city结构为:id int, name varchar 1 bj 2 sz 3 sh 本案例的结果为:select s.id, s.name, s.cityId, c.name from student s

Spark Streaming和Kafka整合开发指南(一)

Apache Kafka是一个分布式的消息发布-订阅系统.可以说,任何实时大数据处理工具缺少与Kafka整合都是不完整的.本文将介绍如何使用Spark Streaming从Kafka中接收数据,这里将会介绍两种方法:(1).使用Receivers和Kafka高层次的API:(2).使用Direct API,这是使用低层次的KafkaAPI,并没有使用到Receivers,是Spark 1.3.0中开始引入的.这两种方法有不同的编程模型,性能特点和语义担保.下文将会一一介绍. 基于Receiver

【转】Spark Streaming和Kafka整合开发指南

基于Receivers的方法 这个方法使用了Receivers来接收数据.Receivers的实现使用到Kafka高层次的消费者API.对于所有的Receivers,接收到的数据将会保存在Spark executors中,然后由Spark Streaming启动的Job来处理这些数据. 然而,在默认的配置下,这种方法在失败的情况下会丢失数据,为了保证零数据丢失,你可以在Spark Streaming中使用WAL日志,这是在Spark 1.2.0才引入的功能,这使得我们可以将接收到的数据保存到WA