Spark Streaming运行kafka数据源

实验内容

了解kafka的基本知识，对kafka进行安装和基础环境配置
2. 安装和准备flume
3. 编译相关测试代码，测试环境，对word count进行spark streaming

Kafka基本介绍
Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群机来提供实时的消费。

Kafka是非常流行的日志采集系统，可以作为DStream的高级数据源，也可以作为Flume的Sink端接收来自Flume的数据。本实验同时用到这两种功能。

Kafka的使用依赖于zookeeper,安装Kafka前必须先安装zookeeper.

实验要求

1. 文档格式工整规范，一目了然。
2. 文字描述与截图说明相结合，清清楚楚。

实验步骤

一、Ubuntu 系统安装Kafka
访问Kafka官方下载页面,下载稳定版本0.10.1.0的kafka.此安装包内已经附带zookeeper,不需要额外安装zookeeper.按顺序执行如下步骤:

1.	cd ~/下载
2.	sudo tar -zxf kafka_2.11-0.10.1.0.tgz -C /usr/local
3.	cd /usr/local
4.	sudo mv kafka_2.11-0.10.1.0/ ./kafka
5.	sudo chown -R hadoop ./kafka

此处我选择的是kafa 2.11-2.0.1.tgz，下载后需要解压：

核心概念
下面介绍Kafka相关概念,以便运行下面实例的同时，更好地理解Kafka.

1.Broker Kafka集群包含一个或多个服务器，这种服务器被称为broker
2. Topic 每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
3. Partition Partition是物理上的概念，每个Topic包含一个或多个Partition.
4. Producer 负责发布消息到Kafka broker
5. Consumer 消息消费者，向Kafka broker读取消息的客户端。
6. Consumer Group 每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）

测试简单实例
接下来在Ubuntu系统环境下测试简单的实例。按顺序执行如下命令：

1.	# 进入kafka所在的目录
2.	cd /usr/local/kafka
3.	bin/zookeeper-server-start.sh config/zookeeper.properties

Shell 命令
命令执行后不会返回Shell命令输入状态,zookeeper就会按照默认的配置文件启动服务,请千万不要关闭当前终端.启动新的终端，输入如下命令：

1.	cd /usr/local/kafka
2.	bin/kafka-server-start.sh config/server.properties
3.	

Shell 命令
kafka服务端就启动了,请千万不要关闭当前终端。启动另外一个终端,输入如下命令:

1.	cd /usr/local/kafka
2.	bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic dblab

Shell 命令
topic是发布消息发布的category,以单节点的配置创建了一个叫dblab的topic.可以用list列出所有创建的topics,来查看刚才创建的主题是否存在。

1.	bin/kafka-topics.sh --list --zookeeper localhost:2181 

Shell 命令
可以在结果中查看到dblab这个topic存在。接下来用producer生产点数据：

1.	bin/kafka-console-producer.sh --broker-list localhost:9092 --topic dblab

Shell 命令
并尝试输入如下信息：

hello hadoop
hello xmu
hadoop world

然后再次开启新的终端或者直接按CTRL+C退出。然后使用consumer来接收数据,输入如下命令：

1.	cd /usr/local/kafka
2.	bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic dblab --from-beginning  

Shell 命令

便可以看到刚才产生的三条信息。说明kafka安装成功。
二、Flume的安装和准备

Flume是Cloudera提供的一个高可用的，高可靠的，分布式的海量日志采集、聚合和传输的系统，Flume支持在日志系统中定制各类数据发送方，用于收集数据；同时，Flume提供对数据进行简单处理，并写到各种数据接受方（可定制）的能力。
Flume主要由3个重要的组件构成：
Source:完成对日志数据的收集，分成transtion 和 event 打入到channel之中。
Channel:主要提供一个队列的功能，对source提供中的数据进行简单的缓存。
Sink:取出Channel中的数据，进行相应的存储文件系统，数据库，或者提交到远程服务器。
Flume逻辑上分三层架构：agent，collector，storage

agent用于采集数据，agent是flume中产生数据流的地方，同时，agent会将产生的数据流传输到collector。
collector的作用是将多个agent的数据汇总后，加载到storage中。
storage是存储系统，可以是一个普通file，也可以是HDFS，HIVE，HBase等。

Flume的架构主要有一下几个核心概念：

Event：一个数据单元，带有一个可选的消息头
Flow：Event从源点到达目的点的迁移的抽象
Client：操作位于源点处的Event，将其发送到Flume Agent
Agent：一个独立的Flume进程，包含组件Source、Channel、Sink
Source：用来消费传递到该组件的Event
Channel：中转Event的一个临时存储，保存有Source组件传递过来的Event
Sink：从Channel中读取并移除Event，将Event传递到Flow Pipeline中的下一个Agent（如果有的话）

flume下载地址： flume下载官网
1.解压安装包

1.	 sudo tar -zxvf apache-flume-1.7.0-bin.tar.gz -C /usr/local # 将apache-flume-1.7.0-bin.tar.gz解压到/usr/local目录下，这里一定要加上-C否则会出现归档找不到的错误
2.	 sudo mv ./apache-flume-1.7.0-bin ./flume   #将解压的文件修改名字为flume，简化操作
3.	 sudo chown -R hadoop:hadoop ./flume  #把/usr/local/flume目录的权限赋予当前登录Linux系统的用户，这里假设是hadoop用户

Shell
2.配置环境变量

1.	  sudo vim ~/.bashrc

Shell
然后在首行加入如下代码：

1.	 export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64;
2.	 export FLUME_HOME=/usr/local/flume                   
3.	 export FLUME_CONF_DIR=$FLUME_HOME/conf
4.	 export PATH=$PATH:$FLUME_HOME/bin

Shell
注意， 上面的JAVA_HOME，如果以前已经在.bashrc文件中设置过，就不要重复添加了，使用以前的设置即可。
比如，以前设置得JAVA_HOME可能是“export JAVA_HOME=/usr/lib/jvm/default-java”，则使用原来的设置即可。
接下来使环境变量生效：

1.	  source ~/.bashrc

Shell
修改 flume-env.sh 配置文件：

1.	cd /usr/local/flume/conf 
2.	sudo cp ./flume-env.sh.template ./flume-env.sh
3.	sudo vim ./flume-env.sh

Shell
打开flume-env.sh文件以后，在文件的最开始位置增加一行内容，用于设置JAVA_HOME变量：
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64;
注意，你的JAVA_HOME可能与上面的设置不一致，一定要根据你之前已经安装的Java路径来设置，比如，有的机器可能是：
export JAVA_HOME=/usr/lib/jvm/default-java
然后，保存flume-env.sh文件，并退出vim编辑器。

3.查看flume版本信息

1.	  cd /usr/local/flume
2.	    ./bin/flume-ng version #查看flume版本信息；

Shell
如果安装成功，出现如下图片

注意：如果系统里安装了hbase，会出现错误: 找不到或无法加载主类 org.apache.flume.tools.GetJavaProperty。如果没有安装hbase，这一步可以略过。

1.	  cd  /usr/local/hbase/conf
2.	  sudo vim hbase-env.sh

1.	  #1、将hbase的hbase.env.sh的这一行配置注释掉,即在export前加一个#
2.	  #export HBASE_CLASSPATH=/home/hadoop/hbase/conf
3.	  #2、或者将HBASE_CLASSPATH改为JAVA_CLASSPATH,配置如下
4.	  export JAVA_CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
5.	  #笔者用的是第一种方法

三、Spark准备工作

要通过Kafka连接Spark来进行Spark Streaming操作，Kafka和Flume等高级输入源，需要依赖独立的库（jar文件）。也就是说Spark需要jar包让Kafka和Spark streaming相连。按照我们前面安装好的Spark版本，这些jar包都不在里面，为了证明这一点，我们现在可以测试一下。请打开一个新的终端，输入以下命令启动spark-shell：

1.	cd /usr/local/spark
2.	./bin/spark-shell

Shell 命令
启动成功后，在spark-shell中执行下面import语句：

1.	import org.apache.spark.streaming.kafka._

Shell 命令
你可以看到，马上会报错如下，因为找不到相关的jar包。

<console>:23: error: object kafka is not a member of package org.apache.spark.streaming
       import org.apache.spark.streaming.kafka._
                                         ^

根据Spark官网的说明，对于Spark2.1.0版本，如果要使用Kafka，则需要下载spark-streaming-kafka-0-8_2.11相关jar包。
现在请在Linux系统中，打开火狐浏览器，请点击这里访问官网，里面有提供spark-streaming-kafka-0-8_2.11-2.1.0jar文件的下载，其中，2.11表示scala的版本，2.1.0表示Spark版本号。下载后的文件会被默认保存在当前Linux登录用户hadoop的下载目录下(“/home/hadoop/下载”)。
我们就把这个文件复制到Spark目录的jars目录下。请新打开一个终端，输入下面命令：

1. cd /usr/local/spark/jars
2. mkdir kafka
3. cd ~
4. cd 下载
5. cp ./spark-streaming-kafka-0-8_2.11-2.1.0.jar /usr/local/spark/jars/kafka

这样，我们就把spark-streaming-kafka-0-8_2.11-2.1.1.jar文件拷贝到了“/usr/local/spark/jars/kafka”目录下。
下面还要继续把Kafka安装目录的libs目录下的所有jar文件复制到“/usr/local/spark/jars/kafka”目录下输入以下命令：

1.	cd /usr/local/kafka/libs
2.	ls
3.	cp ./* /usr/local/spark/jars/kafka

Shell 命令
至此，所有环境准备工作已全部完成，下面开始编写代码。

四、实验过程
1.编写Flume配置文件flume_to_kafka.conf
输入命令：

1.	cd /usr/local/flume
2.	cd conf
3.	vim flume_to_kafka.conf

Shell 命令
内容如下：

a1.sources=r1
a1.channels=c1
a1.sinks=k1
#Describe/configure the source 
a1.sources.r1.type=netcat
a1.sources.r1.bind=localhost
a1.sources.r1.port=33333
#Describe the sink
a1.sinks.k1.type=org.apache.flume.sink.kafka.KafkaSink  
a1.sinks.k1.kafka.topic=test  
a1.sinks.k1.kafka.bootstrap.servers=localhost:9092  
a1.sinks.k1.kafka.producer.acks=1  
a1.sinks.k1.flumeBatchSize=20  
#Use a channel which buffers events in memory  
a1.channels.c1.type=memory
a1.channels.c1.capacity=1000000
a1.channels.c1.transactionCapacity=1000000
#Bind the source and sink to the channel
a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

2.编写Spark Streaming程序(进行词频统计的程序)
首先创建scala代码的目录结构。
输入命令：

1.	cd /usr/local/spark/mycode
2.	mkdir flume_to_kafka
3.	cd flume_to_kafka
4.	mkdir -p src/main/scala
5.	cd src/main/scala

Shell 命令
输入命令：

1.	vim KafkaWordCounter.scala

Shell 命令
KafkaWordCounter.scala是用于单词词频统计，它会把从kafka发送过来的单词进行词频统计，代码内容如下：

package org.apache.spark.examples.streaming
import org.apache.spark._
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.kafka.KafkaUtils

object KafkaWordCounter{
def main(args:Array[String]){
StreamingExamples.setStreamingLogLevels()
val sc=new SparkConf().setAppName("KafkaWordCounter").setMaster("local[2]")
val ssc=new StreamingContext(sc,Seconds(10))
ssc.checkpoint("file:///usr/local/spark/mycode/flume_to_kafka/checkpoint") //设置检查点
val zkQuorum="localhost:2181" //Zookeeper服务器地址
val group="1"  //topic所在的group，可以设置为自己想要的名称，比如不用1，而是val group = "test-consumer-group" 
val topics="test" //topics的名称          
val numThreads=1 //每个topic的分区数
val topicMap=topics.split(",").map((_,numThreads.toInt)).toMap
val lineMap=KafkaUtils.createStream(ssc,zkQuorum,group,topicMap)
val lines=lineMap.map(_._2)
val words=lines.flatMap(_.split(" "))
val pair=words.map(x => (x,1))
val wordCounts=pair.reduceByKeyAndWindow(_ + _,_ - _,Minutes(2),Seconds(10),2) 
wordCounts.print
ssc.start
ssc.awaitTermination
}
}

reduceByKeyAndWindow函数作用解释如下：
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]) 更加高效的reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce”操作。但是，只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”（以InvFunc参数传入）；
此代码中就是一个窗口转换操作reduceByKeyAndWindow，其中，Minutes(2)是滑动窗口长度，Seconds(10)是滑动窗口时间间隔（每隔多长时间滑动一次窗口）。reduceByKeyAndWindow中就使用了加法和减法这两个reduce函数，加法和减法这两种reduce函数都是“可逆的reduce函数”，也就是说，当滑动窗口到达一个新的位置时，原来之前被窗口框住的部分数据离开了窗口，又有新的数据被窗口框住，但是，这时计算窗口内单词的词频时，不需要对当前窗口内的所有单词全部重新执行统计，而是只要把窗口内新增进来的元素，增量加入到统计结果中，把离开窗口的元素从统计结果中减去，这样，就大大提高了统计的效率。尤其对于窗口长度较大时，这种“逆函数”带来的效率的提高是很明显的。
3.创建StreamingExamples.scala
继续在当前目录(/usr/local/spark/mycode/flume_to_kafka/src/main/scala)下创建StreamingExamples.scala代码文件，用于设置log4j:
输入命令：

1.	vim StreamingExamples.scala

Shell 命令
代码内容如下：

package org.apache.spark.examples.streaming
import org.apache.spark.internal.Logging
import org.apache.log4j.{Level, Logger}
//Utility functions for Spark Streaming examples. 
object StreamingExamples extends Logging {
//Set reasonable logging levels for streaming if the user has not configured log4j. 
  def setStreamingLogLevels() {
    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if (!log4jInitialized) {
      // We first log something to initialize Spark's default logging, then we override the
      // logging level.
      logInfo("Setting log level to [WARN] for streaming example." +" To override add a custom log4j.properties to the classpath.")
      Logger.getRootLogger.setLevel(Level.WARN)
    }
  }
}

4.打包文件simple.sbt
输入命令：

1.	cd /usr/local/spark/mycode/flume_to_kafka

Shell 命令
输入命令：

1.	vim simple.sbt
2.	内容如下：
3.	name := "Simple Project"
4.	version := "1.0"
5.	scalaVersion := "2.11.8"
6.	libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
7.	libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.1.0"
8.	libraryDependencies += "org.apache.spark" % "spark-streaming-kafka-0-8_2.11" % "2.1.0"
9.	要注意版本号一定要设置正确。

在/usr/local/spark/mycode/flume_to_kafka目录下输入命令：

1.	cd /usr/local/spark/mycode/flume_to_kafka
2.	find .

Shell 命令
打包之前，这条命令用来查看代码结构，目录结构如下所示：

    .
    ./simple.sbt
    ./src
    ./src/main
    ./src/main/scala
    ./src/main/scala/KafkaWordCounter.scala
    ./src/main/scala/StreamingExamples.scala

5.打包编译
一定要在/usr/local/spark/mycode/flume_to_kafka目录下运行打包命令
输入命令：

1.	cd /usr/local/spark/mycode/flume_to_kafka
2.	/usr/local/sbt/sbt package

Shell 命令
第一次打包的过程可能会很慢，请耐心等待几分钟。打包成功后，会看到如下SUCCESS的提示。

hadoop@dblab-VirtualBox:/usr/local/spark/mycode/flume_to_kafka$ /usr/local/sbt/sbt package
OpenJDK 64-Bit Server VM warning: ignoring option MaxPermSize=256M; support was removed in 8.0
[info] Set current project to Simple Project (in build file:/usr/local/spark/mycode/flume_to_kafka/)
[info] Compiling 2 Scala sources to /usr/local/spark/mycode/flume_to_kafka/target/scala-2.11/classes...
[info] Packaging /usr/local/spark/mycode/flume_to_kafka/target/scala-2.11/simple-project_2.11-1.0.jar ...
[info] Done packaging.
[success] Total time: 7 s, completed 2018-9-5 10:03:08

6.启动zookeeper和kafka
先启动zookeeper
输入命令：

1.	cd /usr/local/kafka
2.	./bin/zookeeper-server-start.sh config/zookeeper.properties

Shell 命令
屏幕显示如下，不要关闭这个终端。

打开第二个终端，然后输入下面命令启动Kafka服务：
输入命令：

1.	cd /usr/local/kafka
2.	bin/kafka-server-start.sh config/server.properties

Shell 命令
屏幕显示如下，不要关闭这个终端

7.运行程序KafkaWordCounter
打开第三个终端，我们已经创建过topic，名为test（这是你之前在flume_to_kafka.conf中设置的topic名字），端口号2181。在第三个终端运行“KafkaWordCounter”程序，进行词频统计，由于现在没有启动输入，所以只有提示信息，没有结果。
输入命令：

1.	cd /usr/local/spark
2.	/usr/local/spark/bin/spark-submit --driver-class-path /usr/local/spark/jars/*:/usr/local/spark/jars/kafka/* --class "org.apache.spark.examples.streaming.KafkaWordCounter" /usr/local/spark/mycode/flume_to_kafka/target/scala-2.11/simple-project_2.11-1.0.jar

Shell 命令
其中”/usr/local/spark/jars/“和”/usr/local/spark/jars/kafka/”用来指明引用的jar包，“org.apache.spark.examples.streaming.KafkaWordCounter”代表包名和类名，这是编写KafkaWordCounter.scala里面的包名和类名，最后一个参数用来说明打包文件的位置。

执行该命令后，屏幕上会显示程序运行的相关信息，并会每隔10秒钟刷新一次信息，用来输出词频统计的结果，此时还只有提示信息，如下所示：

Time: 1536113420000 ms

---

Time: 1536113430000 ms

---

Time: 1536113440000 ms

---

Time: 1536113450000 ms

---

Time: 1536113460000 ms

按Ctrl+z可以退出程序运行
在启动Flume之前，Zookeeper和Kafka要先启动成功，不然启动Flume会报连不上Kafka的错误。
8.启动flume agent
打开第四个终端，在这个新的终端中启动Flume Agent
输入命令：

1.	cd /usr/local/flume
2.	bin/flume-ng agent --conf ./conf --conf-file ./conf/flume_to_kafka.conf --name a1 -Dflume.root.logger=INFO,console

Shell 命令
启动agent以后，该agent就会一直监听localhost的33333端口，这样，我们下面就可以通过“telnet localhost 33333”命令向Flume Source发送消息。这个终端也不要关闭，让它一直处于监听状态。

9.打开第五个终端，发送消息
输入命令：

1.	telnet localhost 33333

遇到了connection refused的问题。

查看netstat状态，无返回值，说明telnet没有启动没有listen。

启动成功。

Shell 命令
这个端口33333是在flume conf文件中设置的source
在这个窗口里面随便敲入若干个字符和若干个回车，这些消息都会被Flume监听到，Flume把消息采集到以后汇集到Sink，然后由Sink发送给Kafka的topic(test)。因为spark Streaming程序不断地在监控topic，在输入终端和前面运行词频统计程序那个终端窗口内看到类似如下的统计结果：
Flume发送之前，KafkaWordCounter检测不到消息，没有统计结果，只有提示信息。
在该窗口输入数据：(每一行结尾加一个空格再按回车发送）

hello dblab 
hello xmu 
hello spark  
大数据 

可以看到统计结果如下，其中“，4）”是对换行符的统计

Time: 1536131370000 ms

(hello,3)
(xmu,1)
,4)
(spark,1)
(大数据,1)
(dblab,1)

输入数据：只输入一个回车
统计结果如下：

Time: 1536131400000 ms

(hello,3)
(xmu,1)
,5)
(spark,1)
(大数据,1)
(dblab,1)

可以看到如上所示的统计结果，其他的都没变，“4）“变成“，5）”，这是换行符的数量加一。
可以看到，统计结果在不断的累积，但是随着运行屏幕显示其实还和窗口设置有关。我试过输完最后一个单词直接敲回车发送，但是接收统计端会把换行符和最后一个单词组合起来形成错误的格式，导致统计结果有误。所以在每行输入完成后先多敲一个空格再按换行符发送数据。
比如输入“hello spark”(spark末尾不加空格)
统计结果如下：

Time: 1536131440000 ms

(hello,4)
(xmu,1)
,1)ark
,5)
(spark,1)
(大数据,1)
(dblab,1)

可以看到”hello”被正确的统计了，数量加一，然而“spark”却和换行符结合变成不能识别的格式“,1)ark”。
所以，这里在发送每一行数据时都要在末尾输入一个空格再按下回车。
运行时应该有5个终端处于开启状态，分别是
Zookeeper服务\Kafka服务\Spark Streaming(KafkaWordCounter)\flume agent\telnet发送端

可以看到另一个终端上会出现：

至此，就完成了Flume、Kafka和Spark Streaming整合进行词频统计的任务。

实验结果及相关结论

运用了Kafka辅助flume这样的海量日志收集工具进行了基础的数据源运行。
验证了Kafka的生产者消费者机制，保证了Kafka的基础运行。

kafka对外使用topic的概念，生产者往topic里写消息，消费者从读消息。为了做到水平扩展，一个topic实际是由多个partition组成的，遇到瓶颈时，可以通过增加partition的数量来进行横向扩容。单个parition内是保证消息有序。每新写一条消息，kafka就是在对应的文件append写，所以性能非常高。