SparkでKafkaを経由して外部のデータを読取りします「Kafka稼働環境作成」

本章では、Kafkaクラスタの稼働環境を作成してみます。
サーバが３台あります。すべてのサーバに、ZookeeperとKafkaを一緒にインストールします。

#１．Zookeeperのインストール
３台のサーバに、Zookeeperをともにインストールします。
個々Zookeeperサーバのインストール方法が大体同じなので、一つのZookeeperサーバを例として説明しておきます。
※Zookeeperの動きは、JavaにベースされているのでJDKを事前にインストールすることが必要です。

**Zookeeperのダウンロード**
Zookeeperをダウンロードして、指定したフォルダーに移動します。

```
# zookeeperダウンロード
cd /usr/work
wget http://ftp.jaist.ac.jp/pub/apache/zookeeper/zookeeper-3.4.13/zookeeper-3.4.13.tar.gz
tar zxvf zookeeper-3.4.13.tar.gz
# 移動
sudo mv zookeeper-3.4.13 /usr/local/lib/zookeeper-3.4.13
cd /usr/local/lib/
sudo ln -s zookeeper-3.4.13 zookeeper
sudo chown -R cluster-kafka kafka
sudo chgrp -R cluster-kafka kafka
```
**Zookeeper環境変数作成**
Zookeeperのサービスを便利に立ち上げるために、環境変数ファイルを作成します。

```
# zookeeper.sh作成
cd /etc/profile.d
sudo vi zookeeper.sh
# 以下の変数を追加、保存
export ZOOKEEPER_HOME=/usr/local/lib/zookeeper
export PATH=$ZOOKEEPER_HOME/bin:$PATH
# 所有者変更
sudo chown cluster-kafka zookeeper.sh
sudo chgrp cluster-kafka zookeeper.sh
# 有効化
source kafka.sh
```
**Zookeeperの配置ファイル編集**
Zookeeperの配置ファイルを編集します。

```
cd /usr/local/lib/zookeeper
cp -rf conf/zoo_sample.cfg conf/zoo.cfg
#　zoo.cfg編集
tickTime=2000
initLimit=5
syncLimit=2
dataDir=/var/lib/zookeeper
server.0=192.168.0.97:2888:3888
server.1=192.168.0.98:2888:3888
server.2=192.168.0.99:2888:3888
```

**配置オプションの使い方**

オプション|解説
:--|:--
tickTime<img width="50px" />|ハートビートを行うために使用され、tickTimeの2倍が最小セッションタイムアウトになる。
initLimit |zkサーバのクオーラムがリーダーに接続しなければならない制限時間。<br>「tickTime * initLimit」より決定
syncLimit |リーダーに同期されるまでの制限時間。「tickTime * syncLimit」より決定
DataDir   |Zookeeperのデータ保存先です。
server.X  |通信に使う接続情報。<br>XはサーバのIDでクラスタ内で各ノードが一意となる数字です.

Zookeeperのデータディレクトリ用意
zoo.cfg中のDataDirに合わせてZookeeperのデータディレクトリを作成します。

```
#　データディレクトリ作成
sudo mkdir /var/lib/zookeeper
sudo chown -R username /var/lib/zookeeper
sudo chgrp -R username /var/lib/zookeeper
```
各ノードが自分のIDが認識できるように、ZookeeperのDataDir直下にmyidファイルを作成します.
myidに記述したIDは、config/zoo.cfgのserver.Xに書いたものによって指定します。

```
#　myid作成
echo "0" > /var/lib/zookeeper/myid     ※１台目
echo "1" > /var/lib/zookeeper/myid     ※２台目
echo "2" > /var/lib/zookeeper/myid     ※３台目
```

**ファイアウォール設定**
クライアントが接続するポート (デフォルトは 2181/tcp) とノード間が通信するポート (前述した 2 つ) が接続できるようにしておきます。

```
sudo firewall-cmd --add-port 2181/tcp --add-port 2888/tcp --add-port 3888/tcp --permanent
sudo firewall-cmd --reload
```

**Zookeeper起動**

```
cd /usr/local/lib/zookeeper
bin/zkServer.sh start
```
ちなみにZookeeperを停止したい場合は、以下を実施してください。

```
bin/zkServer.sh stop
#Zookeeperの稼働状態確認
ps -ef | grep zoo.cfg
```

**leaderかfollowerかの確認方法**

```
cd /usr/local/lib/zookeeper
bin/zkServer.sh status
```
#以下ように表示したら、当該ノードがリーダーになります。
> ZooKeeper JMX enabled by default
> Using config: /usr/local/lib/zookeeper/bin/../conf/zoo.cfg
> Mode: leader

当該ノードがリーダーにならない場合は、以下ように表示します。
> ZooKeeper JMX enabled by default
> Using config: /usr/local/zookeeper/bin/../conf/zoo.cfg
> Mode: follower

#２．Kafkaのインストール
３台のサーバに、Kafkaをともにインストールします。
各サーバのインストール方法が大体同じで、一つのサーバを例として説明しておきます。

```
**Kafkaのダウンロード**
Kafkaをダウンロードして、指定したフォルダーに移動します。
# kafkaダウンロード
cd /usr/work
wget http://ftp.jaist.ac.jp/pub/apache/kafka/2.1.0/kafka_2.11-2.1.0.tgz 
tar zxvf kafka_2.11-2.1.0.tgz
# 移動
sudo mv kafka_2.11-2.1.0 /usr/local/lib/kafka_2.11-2.1.0
cd /usr/local/lib/
sudo ln -s kafka_2.11-2.1.0 kafka
sudo chown -R cluster-kafka kafka
sudo chgrp -R cluster-kafka kafka
```

**Kafkaの環境変数作成**
Kafkaのサービスを便利に立ち上げるために、環境変数ファイルを作成します。

```
# kafka.sh作成
cd /etc/profile.d
sudo vi kafka.sh
# 以下の変数を追加、保存
export KAFKA_HOME=/usr/local/lib/kafka
export PATH=$KAFKA_HOME/bin:$PATH
# 所有者変更
sudo chown cluster-kafka kafka.sh
sudo chgrp cluster-kafka kafka.sh
# 有効化
source kafka.sh
```
**Kafka配置ファイル編集**

```
cd /usr/local/lib/kafka
vi config/server.properties
# 以下の編集確認要
broker.id=0
listeners=PLAINTEXT://192.168.0.99:9092
log.dirs=/tmp/kafka_logs
zookeeper.connect=192.168.0.97:2181,192.168.0.98:2181,192.168.0.99:2181
```
**配置オプションの使い方**

オプション|解説
:--|:--
broker.id<img width="100px" />|クラスタ内のユニークIDです。<br>整数で普通は0からです。
listeners          |当該BrokerのIP、ポートです。
log.dirs           |Kafkaデータの保存先です。
zookeeper.connect  |Zookeeperクラスターリストです。<br>複数台のZookeeperがあれば設定は以下の通りになります。

**Kafkaの起動**

```
cd /usr/local/lib/kafka
bin/kafka-server-start.sh -daemon config/server.properties      
```
※-daemon option がなければブロッキングモードで起動される
ちなみに、Kafkaを停止したい場合は、以下のコマンドを実行してください。

```
bin/kafka-server-stop.sh
```
Kafkaの稼働状態を確認しましょう。

```
ps -ef | grep config/server
```
**Topic新規作成**
いったん、Kafkaを動かして試して見るために、まず、Kafka上にあるトピックを作ってみます。
Kafkaのトピック作成する時には、パーティション数やパーティションの複製となるレプリカの数を指定する必要があります。
ここではパーティション数3、レプリカ数3とします。

```
bin/kafka-topics.sh --create --topic topic-mqtt-kafka --replication-factor 3 --partitions 3 --zookeeper 192.168.0.99:2181
```

**オプションの使い方**

オプション|解説
:--|:--
--zookeeper<img width="120px" />|Zookeeperクラスターリストで、Zookeeperが複数台の場合は、複数のIPをカンマに区切りしてリストように設定します。
--replication-factor |オプションの値が各パーティションのレプリカの数です。<br>各BrokerはTopic毎に1つのレプリカしか置けないので、レプリカの数はBrokerの数以下でなければなりません。
--partitions         |オプションの値が分割するパーティションの数です。<br>パーティションの数は、Brokerの数の倍数で指定した方が良いです。<br>倍数に指定したら均等的にKafkaクラスタ上の全てのBrokerに配布されることができます。

**Topicの稼働状況確認**
以下のコマンドを実行して、すべての稼働中のトピックを確認してみましょう。

```
bin/kafka-topics.sh --describe --zookeeper  192.168.0.99:2181 --topic topic-mqtt-kafka
#以下が表示される
Topic:topic-mqtt-kafka    PartitionCount:3    ReplicationFactor:3 Configs:
Topic: topic-mqtt-kafka   Partition: 0    Leader: 2   Replicas: 2,1,0   Isr: 2,1,0
Topic: topic-mqtt-kafka   Partition: 1    Leader: 1   Replicas: 1,2,0   Isr: 1,2,0
Topic: topic-mqtt-kafka   Partition: 2    Leader: 0   Replicas: 0,1,2   Isr: 0,1,2
```

**トピックの見方**

オプション|解説
:--|:--
Partition<img width="50px" /> |パーティション番号です。<br>今回は、3分割なので、0〜2番になっています。
Leader    |そのパーティションのリーダーとなるBrokerのIDが表示されています。<br>パーティションへのデータの参照・更新は、LeaderとなっているBrokerに対して行われます。
Replicas  |そのパーティションのレプリカが配置されているBrokerのIDが表示されます。<br>今回はレプリカ数を3としているので、各Brokerがすべてのパーティションのレプリカを持っている形になります。
Isr       |そのパーティションのレプリカのうち、データの同期が完了しているレプリカをもっているBrokerのIDが表示されます。


パーティションのデータ更新は、まず、Leaderとなる Brokerに対して行われ、それが各レプリカに反映されていきます。
このレプリケーション処理が完了しているBrokerを、Isr欄では確認できることになります。
今回はレプリカ数が3なので、BrokerのIDも３つ表示されていればすべての Broker でパーティションが同期されていることになります。


**プロデューサーでメッセージを送信**
Kafkaクラスター上に作られたTopicで読み書きをして、動作確認をしてみます。
まずはプロデューサを起動し、メッセージを送ってみます。

```
bin/kafka-console-producer.sh --broker-list 192.168.0.99:9092 --sync --topic topic-mqtt-kafka
> Hello Kafka!
```
※--broker-list : Brokerクラスターリスト
**コンシューマーでメッセージを受信**
新Terminalを開けて、送ったメッセージを受信しましょう。

```
cd /usr/local/lib/kafka
bin/kafka-console-consumer.sh --bootstrap-server 192.168.0.99:9092 --topic topic-mqtt-kafka --from-beginning
Hello Kafka!
```
-----
**&#9734;&#9733;&#9734; 次の文章へ &#9734;&#9733;&#9734;**
[SparkでKafkaを経由して外部のデータを読取りします「KafkaコネクトMQTT」](https://qiita.com/gakuseikai/items/1cef77ac6eb960bb08bd)