Twitter -> (Kinesis | Kafka) -> Storm -> Redis

AWS Kinesis、Apache Kafka、Apache Stormを連携させるサンプルプログラム

以下のようなサンプルプログラムを作りました。

Twitter Streaming API
　-> Producer　-> (AWS Kinesis | Apache Kafka)　-> Apache Storm[WordCounter]
　　-> Redis

20140905 AWS Night in ITHD LT2 from Nobuyuki Matsui

Vagrant

環境構築用のVagrantfile、Berksfile、各種Cookbook

検証した環境

• HostOS

• MacOS X(10.7)
• Oracle VM VirtualBox(4.3.14)
• Berkshelf(3.1.3)
• Vagrant(1.6.3)
• vagrant-omnibus(1.4.1)
• vagrant-vbguest(0.10.0)

インストール

1. VirtualBox、Vagrant、Berkshelfをインストール
2. git clone https://github.com/nmatsui/twitter-kinesis-kafka-storm_vagrant.git
3. berks vendor cookbooks
4. vagrant up --provision

これにより、下記２つのVMが立ち上がり、/etc/hostsが設定されます。

• VM 1: Apache Zookeeper & Apache Kafka用

• Ubuntu Server(14.04.1 LTS)
• Oracle JDK(1.8.0_20)
• Apache Zookeeper(3.3.3) [Apache Kafka install packageに内包されたバージョン]
• Apache Kafka(0.8.1.1)

• VM 2: Apache Storm & Redis用

• Ubuntu Server(14.04.1 LTS)
• Oracle JDK(1.8.0_20)
• SBT(0.13.5)
• Apache Storm(0.9.2-incubating)

WordCounter

Apache Storm用の単語出現頻度をカウントするプログラム

1. AWS Kinesis用のSpout or Apache Kafka用のSpoutを用いてTweetを取得
2. 下記３つのBoltを用いて単語の出現頻度をカウント
3. KinesisとKafkaのデータ構造差異を吸収
4. kuromojiを用いて形態素解析
5. 単語数のカウント
6. 出現数をスコアとして単語をRedisのソート済みセット型に登録

インストール

1. VM 2上でgit clone https://github.com/nmatsui/twitter-kinesis-kafka-storm_wordcounter.git
2. src/main/resources/*.properties.templateをsrc/main/resources/*.propertiesにリネーム
3. AwsCredentials.propertiesにAWSのAPI情報をセット

• もしプロキシ配下で動作させる場合、kinesis.propertiesにプロキシ情報をセット

1. sbt compile

• Scala本体や各種依存ライブラリを取得するため、初回はかなり時間がかかる

1. sbt assembly

依存ライブラリ

• Java

• Oracle JDK(1.8.0_20)
• Scala(2.10.4)
• SBT(0.13.5)

• Library

• Apache Storm(0.9.2-incubating)
• KafkaSpout(0.9.2-incubating)
• KinesisSpout
• Kuromoji(0.7.7)
• Redis Client(2.12)

KinesisSpoutは、awslabsのkinesis-storm-spoutにbuild.sbtを追加し、バグフィックスしたもの（ https://github.com/nmatsui/kinesis-storm-spout ）を利用

Producer

Twitter Streaming APIから取得したTweetをAWS KinesisもしくはApache Kafkaへ連続的に投入するプログラム

1. Twitter Streaming APIから指定したFilterQueryを用いてTweetを連続的に取得
2. AWS KinesisもしくはApache Kafkaへ投入

インストール

1. VM 2上でgit clone https://github.com/nmatsui/twitter-kinesis-kafka-storm_producer.git
2. src/main/resources/*.properties.templateをsrc/main/resources/*.propertiesにリネーム
3. AwsCredentials.propertiesにAWSのAPI情報をセット

• もしプロキシ配下で動作させる場合、kinesis.propertiesとtwitter.propertiesにプロキシ情報をセット

1. sbt compile

• Scala本体や各種依存ライブラリを取得するため、初回はかなり時間がかかる

1. sbt assembly

依存ライブラリ

• Java

• Oracle JDK(1.8.0_20)
• Scala(2.10.4)
• SBT(0.13.5)

• Library

• Twitter4J(4.0.2)
• Apache Kafka Library(0.8.1.1)
• Amazon Kinesis Client Library(1.1.0)

利用方法

AWS Kinesis準備

1. AWS Kinesisに"tweets"という名前のStreamを作成

• 現在のバージョンではUS EAST(N.Virginia)のKinesisにのみ対応

Apache Kafka準備

1. VM 1にログイン
2. Apache Zookeeper起動
3. Apache Kafka起動
4. Apache Kafkaへ"tweets"という名前のtopicを作成

ZookeeperやKafkaの起動手順等は、公式サイトのQuickStartを参照

Redis起動確認

1. redis-cli -h redis --rawでRedisに接続できることを確認

Apache StromのTopology起動

1. java -jar target/scala-2.10/wordcounter.jar (TEST|KAFKA|KINESIS)

• TEST: AWS KinesisやApache Kafkaに接続せず、ランダムな文章をBoltに流すSpoutに接続
• KAFKA: KafkaSpoutに接続
• KINESIS: KinesisSpoutに接続

Producer起動

1. java -jar target/scala-2.10/tweetstreamer.jar (TEST|KAFKA|KINESIS) query_words...

• TEST: AWS KinesisやApache Kafkaに接続せずTwitter Streaming APIから取得したTweetをログ出力するのみ
• KAFKA: Apache KafkaにTweetをputする
• KINESIS: AWS KinesisにTweetをputする

1. query_wordsは、Twitter Streaming APIの"statuses/filter"エンドポイントに渡されるフィルタワード

• 空白区切りで複数の単語を指定可能
• 指定した単語のいづれかが含まれるTweetが抽出される

正しく環境構築できていれば、取得したTweetがログ出力され、Redisに登録される

Redis確認

1. redis-cli -h redis --rawでRedisに接続
2. Redis-CLI上でzrange words 0 -1 withscoresを実行すると、出現頻度の昇順で単語が表示される

Redisの結果をクリアする場合、Redis-CLI上でdel wordsを実行