如何運行KafkaWordCount

157次閱讀

共計 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 行業資訊頻道了解更多相關知識。

正文完