共计 3233 个字符，预计需要花费 9 分钟才能阅读完成。
如何运行 KafkaWordCount，针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。
概要
Spark 应用开发实践性非常强，很多时候可能都会将时间花费在环境的搭建和运行上，如果有一个比较好的指导将会大大的缩短应用开发流程。Spark Streaming 中涉及到和许多第三方程序的整合，源码中的例子如何真正跑起来，文档不是很多也不详细。
下面主要讲述如何运行 KafkaWordCount，这个需要涉及 Kafka 集群的搭建，还是说的越仔细越好。
搭建 Kafka 集群
步骤 1：下载 kafka 0.8.1 及解压
wget https://www.apache.org/dyn/closer.cgi?path=/kafka/0.8.1.1/kafka_2.10-0.8.1.1.tgz
tar zvxf kafka_2.10-0.8.1.1.tgz
cd kafka_2.10-0.8.1.1
步骤 2：启动 zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties
步骤 3：修改配置文件 config/server.properties，添加如下内容
host.name=localhost
# Hostname the broker will advertise to producers and consumers. If not set, it uses the
# value for host.name if configured. Otherwise, it will use the value returned from
# java.net.InetAddress.getCanonicalHostName().
advertised.host.name=localhost
步骤 4：启动 Kafka server
bin/kafka-server-start.sh config/server.properties
步骤 5：创建 topic
bin/kafka-topics.sh –create –zookeeper localhost:2181 –replication-factor 1 –partitions 1 –topic test
检验 topic 创建是否成功
bin/kafka-topics.sh –list –zookeeper localhost:2181
如果正常返回 test
步骤 6：打开 producer，发送消息
bin/kafka-console-producer.sh –broker-list localhost:9092 –topic test
## 启动成功后，输入以下内容测试
This is a message
This is another message
步骤 7：打开 consumer，接收消息
bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic test –from-beginning
### 启动成功后，如果一切正常将会显示 producer 端输入的内容
This is a message
This is another message
运行 KafkaWordCount
KafkaWordCount 源文件位置 examples/src/main/scala/org/apache/spark/examples/streaming/KafkaWordCount.scala
尽管里面有使用说明，见下文，但如果不是事先对 Kafka 有一定的了解的话，决然不知道这些参数是什么意思，也不知道该如何填写。
/**
* Consumes messages from one or more topics in Kafka and does wordcount.
* Usage: KafkaWordCount
* is a list of one or more zookeeper servers that make quorum
* is the name of kafka consumer group
* is a list of one or more kafka topics to consume from
* is the number of threads the kafka consumer should use
*
* Example:
* `$ bin/run-example \
* org.apache.spark.examples.streaming.KafkaWordCount zoo01,zoo02,zoo03 \
* my-consumer-group topic1,topic2 1`
*/
object KafkaWordCount {
def main(args: Array[String]) {
if (args.length 4) {
System.err.println(Usage: KafkaWordCount )
System.exit(1)
}
StreamingExamples.setStreamingLogLevels()
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName( KafkaWordCount)
val ssc = new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint(checkpoint)
val topicpMap = topics.split(,).map((_,numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicpMap).map(_._2)
val words = lines.flatMap(_.split())
val wordCounts = words.map(x = (x, 1L))
.reduceByKeyAndWindow(_ + _, _ – _, Minutes(10), Seconds(2), 2)
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}
来看一看该如何运行 KafkaWordCount
步骤 1：停止运行刚才的 kafka-console-producer 和 kafka-console-consumer
步骤 2：运行 KafkaWordCountProducer
bin/run-example org.apache.spark.examples.streaming.KafkaWordCountProducer localhost:9092 test 3 5
解释一下参数的意思，localhost:9092 表示 producer 的地址和端口, test 表示 topic，3 表示每秒发多少条消息，5 表示每条消息中有几个单词
步骤 3：运行 KafkaWordCount
bin/run-example org.apache.spark.examples.streaming.KafkaWordCount localhost:2181 test-consumer-group test 1
解释一下参数，localhost:2181 表示 zookeeper 的监听地址，test-consumer-group 表示 consumer-group 的名称，必须和 $KAFKA_HOME/config/consumer.properties 中的 group.id 的配置内容一致，test 表示 topic，1 表示线程数。
关于如何运行 KafkaWordCount 问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注丸趣 TV 行业资讯频道了解更多相关知识。
