【版权声明】博客内容由厦门大学数据库实验室拥有版权，未经允许，请勿转载！
[返回Spark教程首页]
Kafka是非常流行的日志采集系统，可以作为DStream的高级数据源。
Kafka的安装和准备工作
关于Kafka的概念和安装方法，请参考厦门大学数据库实验室博客文章《Kafka的安装和简单实例测试》。在安装的时候，要注意，到Kafka官网下载安装文件时，一定要选择和自己电脑上已经安装的scala版本号一致才可以，本教程安装的Spark版本号是1.6.2，scala版本号是2.10，所以，一定要选择Kafka版本号是2.10开头的。比如，到Kafka官网中，可以下载安装文件Kafka_2.10-0.10.1.0，前面的2.10就是支持的scala版本号，后面的0.10.1.0是Kafka自身的版本号。
这里假设你已经根据这篇博客文章安装成功了Kafka。
下面，我们启动Kafka。
请登录Linux系统（本教程统一使用hadoop用户登录），打开一个终端，输入下面命令启动Zookeeper服务：
cd /usr/local/kafka
./bin/zookeeper-server-start.sh config/zookeeper.properties
注意，执行上面命令以后，终端窗口会返回一堆信息，然后就停住不动了，没有回到shell命令提示符状态，这时，千万不要错误认为死机了，而是Zookeeper服务器启动了，正在处于服务状态。所以，千万不要关闭这个终端窗口，一旦关闭，zookeeper服务就停止了，所以，不能关闭这个终端窗口。
请另外打开第二个终端，然后输入下面命令启动Kafka服务：
cd /usr/local/kafka
bin/kafka-server-start.sh config/server.properties
同样，执行上面命令以后，终端窗口会返回一堆信息，然后就停住不动了，没有回到shell命令提示符状态，这时，千万不要错误认为死机了，而是Kafka服务器启动了，正在处于服务状态。所以，千万不要关闭这个终端窗口，一旦关闭，Kafka服务就停止了，所以，不能关闭这个终端窗口。
当然了，还有一种方式是，采用下面加了“&”的命令:
cd /usr/local/kafka
bin/kafka-server-start.sh config/server.properties &
这样，Kafka就会在后台运行，即使你关闭了这个终端，Kafka也会一直在后台运行。不过，这样做，有时候我们往往就忘记了还有Kafa在后台运行，所以，建议暂时不要用&。
下面先测试一下Kafka是否可以正常使用。再另外打开第三个终端，然后输入下面命令创建一个自定义名称为“wordsendertest”的topic（关于什么是topic，请参考厦门大学数据库实验室博客文章《Kafka的安装和简单实例测试》）：
cd /usr/local/kafka
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic wordsendertest
//这个topic叫wordsendertest，2181是zookeeper默认的端口号，partition是topic里面的分区数，replication-factor是备份的数量，在kafka集群中使用，这里单机版就不用备份了
//可以用list列出所有创建的topics,来查看上面创建的topic是否存在
./bin/kafka-topics.sh --list --zookeeper localhost:2181
这个名称为“wordsendertest”的topic，就是专门负责采集发送一些单词的。
下面，我们需要用producer来产生一些数据，请在当前终端内继续输入下面命令：
./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic wordsendertest
上面命令执行后，你就可以在当前终端内用键盘输入一些英文单词，比如我们可以输入：
hello hadoop
hello spark
这些单词就是数据源，这些单词会被Kafka捕捉到以后发送给消费者。我们现在可以启动一个消费者，来查看刚才producer产生的数据。请另外打开第四个终端，输入下面命令：
cd /usr/local/kafka
./bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic wordsendertest --from-beginning
可以看到，屏幕上会显示出如下结果，也就是刚才你在另外一个终端里面输入的内容：
hello hadoop
hello spark
到这里，与Kafka相关的准备工作就顺利结束了。注意，所有这些终端窗口都不要关闭，要继续留着后面使用。
Spark准备工作
Kafka和Flume等高级输入源，需要依赖独立的库（jar文件）。按照我们前面安装好的Spark版本，这些jar包都不在里面，为了证明这一点，我们现在可以测试一下。请打开一个新的终端，然后启动spark-shell：
cd /usr/local/spark
./bin/spark-shell
启动成功后，在spark-shell中执行下面import语句：
scala> import org.apache.spark.streaming.kafka._
<console>:25: error: object kafka is not a member of package org.apache.spark.streaming
import org.apache.spark.streaming.kafka._
^
你可以看到，马上会报错，因为找不到相关的jar包。所以，现在我们就需要下载spark-streaming-kafka_2.10.jar。
现在请在Linux系统中，打开一个火狐浏览器，请点击这里访问Spark官网，里面有提供spark-streaming-kafka_2.10.jar文件的下载，其中，2.10表示scala的版本。这个下载页面会列出spark-streaming-kafka_2.10.jar的很多版本，我们这里选择1.6.2版本（因为本教程安装的Spark版本是1.6.2），你可以点击1.6.2版本的按钮，进入spark-streaming-kafka_2.10-1.6.2.jar的下载页面，点击下载。下载后的文件会被默认保存在当前Linux登录用户的下载目录下，本教程统一使用hadoop用户名登录Linux系统，所以，文件下载后会被保存到“/home/hadoop/下载”目录下面。现在，我们就把这个文件复制到Spark目录的lib目录下。请新打开一个终端，输入下面命令：
cd /usr/local/spark/lib
mkdir kafka
cd ~
cd 下载
cp ./spark-streaming-kafka_2.10-1.6.2.jar /usr/local/spark/lib/kafka
这样，我们就把spark-streaming-kafka_2.10-1.6.2.jar文件拷贝到了“/usr/local/spark/lib/kafka”目录下。
下面开始的过程，让笔者消耗了2个白天和2个黑夜，测试了网络上各种解决方案，反复失败，最终，只能自己进行“蛮力测试”，下载各种版本进行尝试，最终，得以解决，解决的秘诀是，第一个秘诀是，凡是遇到“java.lang.NoSuchMethodError”这种错误，一定是由版本不一致导致的，也就是spark、scala、spark streaming、Kafka这几个软件的版本不一致，所以，必须保证它们版本的一致性。比如，笔者电脑安装了Spark1.6.2版本，它包含的scala版本是2.10.5，那么，与之对应的straming的版本也必须是spark-streaming_2.10-1.6.2.jar，对应的Kafka也必须是spark-streaming-kafka_2.10-1.6.2.jar。第二个秘诀是，凡是遇到“Class Not Found”这种错误，一般是由缺少jar包引起的。到底缺少什么jar包，笔者在网络上找不到解决方案，也只能靠自己“蒙”，所以，下面笔者测试通过的方法，只是自己反复测试两天后“蒙”对了，但是，也说不出是什么道理。请你按照下面操作即可。
还需要在Linux系统中，打开火狐浏览器，到网络上下载另外一个文件spark-streaming_2.10-1.6.2.jar（下载地址），其中，2.10是scala版本号，1.6.2是Spark版本号。
spark-streaming_2.10-1.6.1.jar下载成功以后是被放到当前Linux登录用户的下载目录下，本教程统一使用hadoop用户名登录Linux系统，所以，文件下载后会被保存到“/home/hadoop/下载”目录下面。现在，我们就把这个文件复制到Spark目录的lib目录下。请新打开一个终端，输入下面命令：
cd /usr/local/spark/lib
cd ~
cd 下载
cp ./spark-streaming_2.10-1.6.1.jar /usr/local/spark/lib/kafka
还没完，下面还要继续把Kafka安装目录的lib目录下的所有jar文件复制到“/usr/local/spark/lib/kafka”目录下，请在终端中执行下面命令：
cd /usr/local/kafka/libs
ls
cp ./* /usr/local/spark/lib/kafka
为什么要拷贝所有的jar过来，因为笔者遇到错误以后，知道是因为缺少jar包，但是，实在不知道是缺少什么jar包，所以，就全部拷贝过来了。但是，全部拷贝过来以后，有些jar包会和spark中已经有的jar包发生冲突，程序一运行就会出现一堆的错误，根据错误提示信息，笔者大概猜测是因为冲突引起的，所以，就删除一些可能引起冲突的jar包，删除一些后，再测试程序，再出错，再删除一批，最后测试通过了，就发现，需要把下面这些jar包删除掉，这样才不会出现错误：
cd /usr/local/kafka/libs
ls
rm log4j*
rm jackson*
上面这些方法纯粹是靠蛮力反复测试才成功的，没有找到问题的本质，也没有网络资料可以帮笔者解决。但是，起码通过这种方式，程序可以顺利运行了。
现在，我们需要配置spark-env.sh文件，让spark能够在启动的时候找到spark-streaming-kafka_2.10-1.6.2.jar等5个jar文件。命令如下：
cd /usr/local/spark/conf
vim spark-env.sh
使用vim编辑器打开了spark-env.sh文件，因为这个文件之前已经反复修改过，目前里面的前面几行的内容应该是这样的：
export SPARK_CLASSPATH=$SPARK_CLASSPATH:/usr/local/spark/lib/hbase/*
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
我们只要简单修改一下，把“/usr/local/spark/lib/kafka/*“增加进去，修改后的内容如下：
export SPARK_CLASSPATH=$SPARK_CLASSPATH:/usr/local/spark/lib/hbase/*:/usr/local/spark/lib/kafka/*
export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
保存该文件后，退出vim编辑器。然后，就可以启动spark-shell：
cd /usr/local/spark
./bin/spark-shell
启动成功后，再次执行命令：
scala> import org.apache.spark.streaming.kafka._
//会显示下面信息
import org.apache.spark.streaming.kafka._
说明导入成功了。这样，我们就已经准备好了Spark环境，它可以支持kafka相关编程了。
编写Spark程序使用Kafka数据源
下面，我们就可以进行程序编写了。请新打开一个终端，然后，执行命令创建代码目录：
cd /usr/local/spark/mycode
mkdir kafka
cd kafka
mkdir -p src/main/scala
cd src/main/scala
vim KafkaWordProducer.scala
使用vim编辑器新建了KafkaWordProducer.scala，它是产生一系列字符串的程序，会产生随机的整数序列，每个整数被当做一个单词，提供给KafkaWordCount程序去进行词频统计。请在KafkaWordProducer.scala中输入以下代码：
import java.util.HashMap
import org.apache.kafka.clients.producer.{ProducerConfig, KafkaProducer, ProducerRecord}
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.SparkConf
object KafkaWordProducer {
def main(args: Array[String]) {
if (args.length < 4) {
System.err.println("Usage: KafkaWordCountProducer <metadataBrokerList> <topic> " +
"<messagesPerSec> <wordsPerMessage>")
System.exit(1)
}
val Array(brokers, topic, messagesPerSec, wordsPerMessage) = args
// Zookeeper connection properties
val props = new HashMap[String, Object]()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer")
val producer = new KafkaProducer[String, String](props)
// Send some messages
while(true) {
(1 to messagesPerSec.toInt).foreach { messageNum =>
val str = (1 to wordsPerMessage.toInt).map(x => scala.util.Random.nextInt(10).toString)
.mkString(" ")
val message = new ProducerRecord[String, String](topic, null, str)
producer.send(message)
}
Thread.sleep(1000)
}
}
}
保存后退出vim编辑器。然后，继续在当前目录下创建KafkaWordCount.scala代码文件：
vim KafkaWordCount.scala
KafkaWordCount.scala是用于单词词频统计，它会把KafkaWordProducer发送过来的单词进行词频统计，代码内容如下：
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 KafkaWordCount{
def main(args:Array[String]){
StreamingExamples.setStreamingLogLevels()
val sc = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
val ssc = new StreamingContext(sc,Seconds(10))
ssc.checkpoint("file:///usr/local/spark/mycode/kafka/checkpoint") //设置检查点，如果存放在HDFS上面，则写成类似ssc.checkpoint("/user/hadoop/checkpoint")这种形式，但是，要启动hadoop
val zkQuorum = "localhost:2181" //Zookeeper服务器地址
val group = "1" //topic所在的group，可以设置为自己想要的名称，比如不用1，而是val group = "test-consumer-group"
val topics = "wordsender" //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
}
}
保存后退出vim编辑器。然后，继续在当前目录下创建StreamingExamples.scala代码文件：
vim StreamingExamples.scala
下面是StreamingExamples.scala的代码，用于设置log4j:
import org.apache.spark.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)
}
}
}
这样，我们在“/usr/local/spark/mycode/kafka/src/main/scala”目录下，就有了如下三个代码文件：
KafkaWordProducer.scala
KafkaWordCount.scala
StreamingExamples.scala
然后，请执行下面命令：
cd /usr/local/spark/mycode/kafka/
vim simple.sbt
在simple.sbt中输入以下代码：
name := "Simple Project"
version := "1.0"
scalaVersion := "2.10.5"
libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.2"
libraryDependencies += "org.apache.spark" % "spark-streaming_2.10" % "1.6.2"
libraryDependencies += "org.apache.spark" % "spark-streaming-kafka_2.10" % "1.6.2"
保存文件退出vim编辑器。然后执行下面命令，进行打包编译：
cd /usr/local/spark/mycode/kafka/
/usr/local/sbt/sbt package
打包成功后，就可以执行程序测试效果了。
首先，请启动hadoop（有可能你前面采用了ssc.checkpoint("/user/hadoop/checkpoint")这种形式，写入HDFS）：
cd /usr/local/hadoop
./sbin/start-all.sh
启动hadoop成功以后，就可以测试我们刚才生成的词频统计程序了。
要注意，我们之前已经启动了zookeeper服务，启动了kafka服务，因为我们之前那些终端窗口都没有关闭，所以，这些服务都在运行。如果你不小心关闭了之前的终端窗口，那就请回到本文前面，再启动zookeeper服务，启动kafka服务。
首先，请新打开一个终端，执行如下命令，运行“KafkaWordProducer”程序，生成一些单词（是一堆整数形式的单词）：
cd /usr/local/spark
/usr/local/spark/bin/spark-submit --class "KafkaWordProducer" /usr/local/spark/mycode/kafka/target/scala-2.10/simple-project_2.10-1.0.jar localhost:9092 wordsender 3 5
注意，上面命令中，"localhost:9092 wordsender 3 5"是提供给KafkaWordProducer程序的4个输入参数，第1个参数localhost:9092是Kafka的broker的地址，第2个参数wordsender是topic的名称，我们在KafkaWordCount.scala代码中已经把topic名称写死掉，所以，KafkaWordCount程序只能接收名称为"wordsender"的topic。第3个参数“3”表示每秒发送3条消息，第4个参数“5”表示，每条消息包含5个单词（实际上就是5个整数）。
执行上面命令后，屏幕上会不断滚动出现新的单词，如下：
3 3 6 3 4
9 4 0 8 1
0 3 3 9 3
0 8 4 0 9
8 7 2 9 5
2 6 4 8 5
0 9 6 0 9
4 0 0 8 1
1 8 3 7 4
4 0 6 5 7
3 9 1 5 0
9 3 9 6 7
1 8 7 4 3
9 5 6 2 6
4 8 8 6 8
0 0 3 3 7
这个终端窗口就放在这里，不要关闭，千万不要关闭，就让它一直不断发送单词。
然后，请新打开一个终端，执行下面命令，运行KafkaWordCount程序，执行词频统计：
cd /usr/local/spark
/usr/local/spark/bin/spark-submit --class "KafkaWordCount" /usr/local/spark/mycode/kafka/target/scala-2.10/simple-project_2.10-1.0.jar
运行上面命令以后，就启动了词频统计功能，屏幕上就会显示如下信息：
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/usr/local/spark/lib/kafka/slf4j-log4j12-1.7.21.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/usr/local/hadoop/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
-------------------------------------------
Time: 1479789000000 ms
-------------------------------------------
(4,16)
(8,14)
(6,15)
(0,10)
(2,9)
(7,17)
(5,14)
(9,9)
(3,8)
(1,8)
恭喜你，顺利完成了Spark Streaming和Kafka的集成。
