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

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

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

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

![TweetStreamWordCounter.png](https://qiita-image-store.s3.amazonaws.com/0/37648/066921d3-ea4d-f264-947d-5a8df2cd82cc.png)

<iframe src="//www.slideshare.net/slideshow/embed_code/38740780?startSlide=18" width="425" height="355" frameborder="0" marginwidth="0" marginheight="0" scrolling="no" style="border:1px solid #CCC; border-width:1px; margin-bottom:5px; max-width: 100%;" allowfullscreen> </iframe> <div style="margin-bottom:5px"> <strong> <a href="//www.slideshare.net/nbykmatsui/20140905-aws-nightinithdlt2/18" title="20140905 AWS Night in ITHD LT2" target="_blank">20140905 AWS Night in ITHD LT2</a> </strong> from <strong><a href="//www.slideshare.net/nbykmatsui" target="_blank">Nobuyuki Matsui</a></strong> </div>

## Vagrant

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

https://github.com/nmatsui/twitter-kinesis-kafka-storm_vagrant

### 検証した環境

* 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用の単語出現頻度をカウントするプログラム

https://github.com/nmatsui/twitter-kinesis-kafka-storm_wordcounter

1. AWS Kinesis用のSpout or Apache Kafka用のSpoutを用いてTweetを取得
1. 下記３つのBoltを用いて単語の出現頻度をカウント
 1. KinesisとKafkaのデータ構造差異を吸収
 2. kuromojiを用いて形態素解析
 3. 単語数のカウント
1. 出現数をスコアとして単語を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](https://github.com/awslabs/kinesis-storm-spout)にbuild.sbtを追加し、バグフィックスしたもの（ https://github.com/nmatsui/kinesis-storm-spout ）を利用

## Producer

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

https://github.com/nmatsui/twitter-kinesis-kafka-storm_producer

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にログイン
1. Apache Zookeeper起動
2. Apache Kafka起動
3. Apache Kafkaへ"tweets"という名前のtopicを作成

ZookeeperやKafkaの起動手順等は、[公式サイトのQuickStart](http://kafka.apache.org/documentation.html#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する
3. `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`を実行