Apache Kafkaの導入〜トピックの生成〜メッセージ送受信(完全版)

Apache Kafka

前提：Zookeeperがポート番号2181番で起動している状態ではじめること。

• Apache Kafkaの導入からトピックの生成、メッセージ送受信方法を公開します。
• Pub/Subに興味があ流けど、Web上の情報だけではうまくいかなかった方、このマニュアルで、環境をそろえた上で、やってみてください。(動作確認済みです。)

zkServer.sh start

1 Scalaをインストールする。

※Scalaのバージョンは、Apache Kafka公式サイトの、推奨バージョンをインストールすること。
※今回は、kafka_2.10-0.8.2の推奨バージョンである、Scala 2.10をインストールする。

su
yum -y install wget
yum -y install tar
yum -y install java-1.6.0-openjdk-devel
mkdir /usr/local/download/
cd /usr/local/download/
wget http://www.scala-lang.org/files/archive/scala-2.10.4.tgz
tar xzvf scala-2.10.4.tgz
mv /usr/local/download/scala-2.10.4 /usr/local/scala-2.10.4
cd /usr/local
ln -s scala-2.10.4/ scala
echo "export PATH=\$PATH:/usr/local/scala/bin" >> /etc/profile;
export PATH=$PATH:/usr/local/scala/bin

2 ダウンロードおよび解凍

cd /usr/local/download
wget http://ftp.meisei-u.ac.jp/mirror/apache/dist/kafka/0.8.2-beta/kafka_2.10-0.8.2-beta.tgz
tar xzvf kafka_2.10-0.8.2-beta.tgz
mv /usr/local/download/kafka_2.10-0.8.2-beta /usr/local/kafka_2.10-0.8.2-beta
cd /usr/local/
ln -s kafka_2.10-0.8.2-beta/ kafka
echo "export PATH=\$PATH:/usr/local/kafka/bin" >> /etc/profile;
export PATH=$PATH:/usr/local/kafka/bin

3 実行

kafka-server-start.sh /usr/local/kafka/config/server.properties &

4 トピックの生成

Let's create a topic named "test" with a single partition and only one replica:

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test2
kafka-topics.sh --list --zookeeper localhost:2181
# >test2

5 メッセージ送受信テスト

5.1 Kafkaのメッセージ送受信をテストします。ターミナル(端末)を2つ起動し、2つともvm1にsshで接続し、root権限にします。

5.1.1 vm1のconsole1で次のコマンドを入力します。→入力待ち状態へ移行します。(プロデューサーとなる。)

kafka-console-producer.sh --broker-list localhost:9092 --topic test2
# >

5.1.2 vm1のconsole2で次のコマンドを入力します。→出力待ち状態へ移行します。(コンシューマーとなる。こちら側のコンソールは表示専用)

kafka-console-consumer.sh --zookeeper localhost:2181 --topic test2 --from-beginning
# >

5.1.3 vm1のconsole1で何かメッセージを入力して、Enter(改行)します。→console2の画面にも、結果が出力されます。(通信成功)

Hello
# >Hello

6 マルチブローカークラスターのための設定

6.1 server.propertiesファイルの編集

• broker.idの設定(Zookeeperと同じように、各vmにユニークなidを振る必要があります。)
• 待ち受けポート番号の設定(VM1のみ9092番、VM2〜VM5は9093番〜9096番に設定します。)
• logディレクトリの設定(/tmp/kafka-logs)
• /tmp/kafka-logsディレクトリの作成

vm1の設定

sed -i -e 's/^broker\.id/#broker\.id/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#broker\.id/a broker\.id=1' /usr/local/kafka/config/server.properties;
sed -i -e 's/^port/#port/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#port/a port=9092' /usr/local/kafka/config/server.properties;
sed -i -e 's/^log\.dir/#log\.dir/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#log\.dir/a log\.dir=\/tmp\/kafka-logs' /usr/local/kafka/config/server.properties;
mkdir /tmp/kafka-logs

vm2の設定

sed -i -e 's/^broker\.id/#broker\.id/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#broker\.id/a broker\.id=2' /usr/local/kafka/config/server.properties;
sed -i -e 's/^port/#port/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#port/a port=9093' /usr/local/kafka/config/server.properties;
sed -i -e 's/^log\.dir/#log\.dir/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#log\.dir/a log\.dir=\/tmp\/kafka-logs' /usr/local/kafka/config/server.properties;
mkdir /tmp/kafka-logs

vm3の設定

sed -i -e 's/^broker\.id/#broker\.id/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#broker\.id/a broker\.id=3' /usr/local/kafka/config/server.properties;
sed -i -e 's/^port/#port/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#port/a port=9094' /usr/local/kafka/config/server.properties;
sed -i -e 's/^log\.dir/#log\.dir/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#log\.dir/a log\.dir=\/tmp\/kafka-logs' /usr/local/kafka/config/server.properties;
mkdir /tmp/kafka-logs

vm4の設定

sed -i -e 's/^broker\.id/#broker\.id/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#broker\.id/a broker\.id=4' /usr/local/kafka/config/server.properties;
sed -i -e 's/^port/#port/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#port/a port=9095' /usr/local/kafka/config/server.properties;
sed -i -e 's/^log\.dir/#log\.dir/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#log\.dir/a log\.dir=\/tmp\/kafka-logs' /usr/local/kafka/config/server.properties;
mkdir /tmp/kafka-logs

vm5の設定

sed -i -e 's/^broker\.id/#broker\.id/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#broker\.id/a broker\.id=5' /usr/local/kafka/config/server.properties;
sed -i -e 's/^port/#port/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#port/a port=9096' /usr/local/kafka/config/server.properties;
sed -i -e 's/^log\.dir/#log\.dir/g' /usr/local/kafka/config/server.properties;
sed -i -e '/^#log\.dir/a log\.dir=\/tmp\/kafka-logs' /usr/local/kafka/config/server.properties;
mkdir /tmp/kafka-logs

7 マルチブローカーでのkafkaの起動

kafka-server-stop.sh
kafka-server-start.sh /usr/local/kafka/config/server.properties &

8 マルチブローカー上でのトピックの作成

kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic my-replicated-topic

9 マルチブローカー上でのトピックの表示

kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic
# >Topic:my-replicated-topic	PartitionCount:1	ReplicationFactor:3	Configs:
# >Topic: my-replicated-topic	Partition: 0	Leader: 2	Replicas: 2,1,3	Isr: 2,1,3

※各項目の説明

• "leader" is the node responsible for all reads and writes for the given partition. Each node will be the leader for a randomly selected portion of the partitions.
• "replicas" is the list of nodes that replicate the log for this partition regardless of whether they are the leader or even if they are currently alive.
• "isr" is the set of "in-sync" replicas. This is the subset of the replicas list that is currently alive and caught-up to the leader.

10 プロデューサーとコンシューマの起動

※起動は数秒置きながら行うとエラーが発生しない。

vm1上でプロデューサーを起動します。

kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

vm2〜vm5でコンシューマを起動します。

kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic

起動がうまくいかない場合

• server.propertiesのhost.nameがコメントアウトされている場合は外す。

host.name=localhost

• logの読み込みに失敗している場合

1. ログファイルを削除します。

rm -rf /var/log/kafka
mkdir /var/log/kafka

1. 手動で起動します。

cd /usr/local/kafka/
pidproxy /var/run/kafka.pid /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.propertie