良く知りもせずKafkaの入門部分を触ってみた #Java

背景

ビッグデータとかリアルタイム解析とかデータマイニングとか機械学習とか、皆が面白そうにブログ書いてたり話してたりしてるから興味あるんですよ。面白そうなんですよ。
で、個人的に興味があって調べていますというただそれだけです。
下手の横好き以下なので、何をいまさらレベルのことをやりながら楽しんでます。

参考

Kindle書籍でこんなのがあって、GWくらいに買ったはずなのにiPadの中で肥やしになっていたので夏休みの自由研究がてら触ってみようかと。ほら日本語ですよ日本語。
Apache Kafka入門@kindle

環境

Ubuntu 14.04(32bit)
OracleJDK 1.6.0_45
Apache Kafka 0.7.2

インストール

公式サイトのパッケージを展開するだけでした。
最初にJDK1.8でやろうとしたけど、0.7x系では動きませんでした。
複数バージョンのJDKを切り替える方法を知らなかったので思わぬところで勉強になった。

JDKのインストール

$ sudo apt-get install -y oracle-java6-installer  
$ sudo apt-get install -y oracle-java7-installer  
$ sudo apt-get install -y oracle-java8-installer

JDKの切り替え

$ update-java-alternatives -l  # 一覧の取得
$ sudo update-java-alternatives -s java-7-oracle # 切り替え

調べれば情報が出てきます。
[http://xyk.hatenablog.com/entry/2013/10/15/175623:title]

構成概要

ブローカは、メッセージをキューの形式で保存し返すことができるサーバ。
プロデューサは、ブローカにメッセージを突っ込む人。
コンシューマは、ブローカに入ってるメッセージを参照する人。Pull型でアクセスする。

ブローカは指定されたサイズ以上のデータを保持したら古い順にメッセージを削除していく。
メッセージは自由フォーマットなのでアプリケーション間で取り決めしておきましょう。
1行表記のUTF-8のJSONとかに決めてしまえば楽なのかな？まだ良く分からないけど。

使ってみよう

とりあえず使う分にはそんなに難しくなかったです。
考えるのは、Brocker(ブローカ)、Producer(プロデューサ)、Consumer(コンシューマ)の３要素とおまけのZookeeperです。

インストール

公式サイトからダウンロードしてきてフォルダまるごと解凍してどこかに置くだけ。

我が家の環境では

$KAFKA_HOME/bin/kafka_run_class.sh

に記載されている-XX:+UseCompressedOopsオプションを削除しないと動かなかったけど、0.8のことだか0.7のことだか忘れました。

なんかJVMがサポートしてないとかなんとかで怒られてました。

Zookeeperを起動する

デフォルトではZookeepeを使用するオプションがONになっているので起動します。

$ cd $KAFKA_HOME
$ bin/zookeeper-server-start.sh config/zookeeper.properties

Brokerサーバを起動する

Zookeeperと連携して動き始めます。逆順にやるとProducerを起動したら死にました。

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

Producerを起動する

体験版用なコンソール用のProducerが入っているのでそちらを起動します。

$ bin/kafka-console-producer.sh --zookeeper localhost:2181 --topic

Consumerを起動する

Consumerにもコンソール用があるので起動します。

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

動作確認

で、Producerのコンソール画面で適当に文字列を入力してみると、

[2014-08-17 13:04:00,752] INFO Creating async producer for broker id = 0 at 127.0.1.1:9092 (kafka.producer.ProducerPool)
わっはっは
[2014-08-17 13:04:25,861] INFO Connected to 127.0.1.1:9092 for producing (kafka.producer.SyncProducer)
われわれはうちゅうじんだ

Consumerの画面に応答が返ってくる。ただそれだけです。

[2014-08-17 13:04:17,385] INFO console-consumer-79113_fjkei-XS35V3-1408248256126-96c479ba Not creating event watcher for trivial whitelist test (kafka.consumer.ZookeeperConsumerConnector)
わっはっは
われわれはうちゅうじんだ

最初のメッセージよりもconnectedのログが遅れてくるのが気持ち悪いけどまあいいや。
さらに、Consumerを再起動すると、下のように過去入力したメッセージを全部取得出来るので、データは消滅しないご様子です。

わっはっは
われわれはうちゅうじんだ

ここまでで基本的な使い方は終わり。

Javaでの実装の仕方とかも書いてあったけど一緒に書くと量ばっかり多くなるからパスします。

公式サイトのサンプルもコピペ十分で動くので、エラーログを読むことが出来れば四苦八苦しながら書けるでしょう。

感想

３日ほどかけて、とりあえずの動作は理解出来ました。

公式ドキュメントのQuickStartがすんなり動いてくれたので非常にやりやすかったです。

Javaの勝手が分からないのでHelloWorldProducerのビルドの仕方が分からず戸惑ってた時間が一番長かったかな？

で、Kafka自体は単体で扱うとただのメッセージキュー管理用のソフトですね。

オフセットとトピックを指定しながら欲しい分データを取得していくのが普通の使い方だと思います。

ここまでの内容だとありがたみがイマイチ分からないので、自分で似たソフトを作っても大差ない気がしてしまうのが難点。

本を読んでいる感じだと、Zookeeperによる連携とかログのキューイングの仕方とかに工夫があって、
容易にスケールさせることができるのが強みなんだろうなと思ってます。

その他

一緒にStormやらFluentdやらと並んでいるのが多いのでその辺りの事情を知っておいた方が世の中の利用方法にあっていそうな感じです。
また、最近はElasticSearch+Kibanaがわっしょいされてる風に見えるので、その辺と連携したせくしーな使い方が出来れば良いなとも。
あと何気にPython3向けのドライバがまだ無いことが個人的には問題。