E.1   安裝Kafka

E.1.1    下載Kafka

Kafka是由LinkedIn設(shè)計的一個高吞吐量、分布式、基于發(fā)布訂閱模式的消息系統(tǒng)，使用Scala編寫，它以可水平擴展、可靠性、異步通信和高吞吐率等特性而被廣泛使用。目前越來越多的開源分布式處理系統(tǒng)都支持與Kafka集成，其中Spark Streaming作為后端流引擎配合Kafka作為前端消息系統(tǒng)正成為當(dāng)前流處理系統(tǒng)的主流架構(gòu)之一。

Kafka安裝包可以在其官網(wǎng)下載頁面下載，下載地址如下，為加快下載速度可以選擇中國境內(nèi)的鏡像，選擇穩(wěn)定版本kafka_2.11-0.9.0.1.tgz安裝包。

http://kafka.apache.org/downloads.html

E.1.2    解壓并配置環(huán)境變量

下載后把安裝包方放在目錄/home/spark/work目錄下，用下面命令解壓縮Kafka安裝包，并把解壓后的目錄移動到/app/soft目錄下：

$cd /home/spark/work/

$tar -zxf kafka_2.11-0.9.0.1.tgz

$mv kafka-0.9.0.1 /app/soft

$ll /app/soft

為了方便運行Kafka相關(guān)腳本，將Kafka的bin路徑加入到/etc/profile中，設(shè)置如下內(nèi)容（分發(fā)到各節(jié)點后，在各節(jié)點上做同樣設(shè)置）：

export KAFKA_HOME=/app/soft/kafka-0.9.0.1

export PATH=$PATH:$KAFKA_HOME/bin

設(shè)置完畢后使用如下命令使配置生效：

$source /etc/profile

E.1.3    修改Kafka的配置文件

在Kafka的根目錄下建立log目錄用于存放日志文件：

$mkdir /app/soft/kafka-0.9.0.1/logs/

修改$KAFKA_HOME/config/server.properties配置文件內(nèi)容（僅列出重要配置）：

################ Server Basics ################

#建議根據(jù)ip區(qū)分,這里使用ZooKeeper中的id來設(shè)置，如master節(jié)點設(shè)置為0，slave1節(jié)點設(shè)置為1，slave2節(jié)點設(shè)置為2

broker.id=1

 

################ Socket Server Settings ################

#broker用于接收producer消息的端口

port=9092

 

#broker的hostname

host.name=master

 

#配置PRODUCER/CONSUMER連上來的時候使用的地址

advertised.host.name=master

 

################ Log Basics ################

#kafka存放消息文件的路徑

log.dirs=/app/soft/kafka-0.9.0.1/logs/

 

#topic的默認(rèn)分區(qū)數(shù)

num.partitions=2

 

################ ZooKeeper ################

#ZooKeeper集群連接地址信息

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

 

#連接ZooKeeper超時時間，單位為毫秒

zookeeper.connection.timeout.ms=6000

E.1.4    分發(fā)Kafka到各節(jié)點

使用scp命令到Kafka分發(fā)到slave1和slave2節(jié)點上：

$cd /app/soft/kafka-0.9.0.1

$scp -r kafka-0.9.0.1 spark@slave1:/app/soft

$scp -r kafka-0.9.0.1 spark@slave2:/app/soft

分發(fā)完畢后，修改server.properties配置文件中broker.id、host.name、advertised.host. name和zookeeper.connect等配置項。

E.2   啟動并驗證

1.  啟動ZooKeeper

分別在master、slave1和slave2節(jié)點上啟動ZooKeeper服務(wù)：

$zkServer.sh start

2.  啟動Kafka

分別在master、slave1和slave2節(jié)點上啟動Kafka服務(wù)：

$kafka-server-start.sh $KAFKA_HOME/config/server.properties

圖 附錄E?1 啟動Kafka

3.  在master節(jié)點上新建主題Topic

$kafka-topics.sh --create --topic kafkaTopic --replication-factor 3 --partitions 2 --zookeeper master:2181

圖 附錄E?2 在Kafka中創(chuàng)建主題

4.  在slave1模擬消息生產(chǎn)者，發(fā)送消息至Kafka

$kafka-console-producer.sh --broker-list master:9092 --sync --topic kafkaTopic

當(dāng)消費者連接后，在發(fā)送消息的終端輸入：hello kafka/who are you?

圖 附錄E?3 slave1模擬消息生產(chǎn)者發(fā)送消息

5.  在slave2模擬消息消費者，顯示消息的消費

$kafka-console-consumer.sh --zookeeper master:2181 --topic kafkaTopic --from-beginning

由于設(shè)置接收從開始到現(xiàn)在的消息，以前發(fā)送的消息也顯示在slave2終端上。

圖 附錄E?4 slave2模擬消息消費者收到消息