Flumeを使いQuickstart VM上でTwitterのツイートを取得する

以下の記事に感化されて、Quickstart VMでツイートの取得環境を構築してみました。正味５分ぐらいで構築できます。

書いた / Flumeを使いMac上でTwitterのツイートを取得する on @Qiita http://t.co/SiJbgYuyqN

— Sho Shimauchi (@shiumachi) May 26, 2015

ま、一言で言えばパクリですw

Cloudera Quickstart VMにはFlume 1.5が含まれているため、Twitterから簡単にデータを取得することができます。詳細は@shiumachi氏のqiitaの記事を参照してください。

準備

Cloudera Quickstart VMにインストールされているFlumeのバージョンを確認します。
Cloudera Managerにログインし、「ホスト」をクリックし、続いて「Host Inspector」をクリックします。

しばらくすると、各ホスト（といっても今回は１台）からパッケージのバージョンなどが表示されます。間違った情報があれば示してくれます。Flumeのバージョンは1.5.0でした。
avrot-toolsはインストールされていたので、Quickstart VM環境では不要です。

Twitter APIキーとアクセストークンの取得

略

Flumeの設定を登録

Cloudera ManagerはFlumeの設定をブラウザから入力できるので、設定ファイルの記述は不要です。また、元の記事ではツイートのログをMacのローカルファイルに保存していましたが、Quickstart VMではHDFSが動作しているので、HDFSに保存先を変更しました。

変更したFlumeの設定

[code]
# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source
a1.sources.r1.type = org.apache.flume.source.twitter.TwitterSource
a1.sources.r1.channels = c1
a1.sources.r1.consumerKey = CONSUMER_KEYを指定
a1.sources.r1.consumerSecret = CONSUMER_KEY_SECRETを指定
a1.sources.r1.accessToken = ACCESS_TOKENを指定
a1.sources.r1.accessTokenSecret = ACCESS_TOKEN_SECRETを指定
a1.sources.r1.maxBatchSize = 50000
a1.sources.r1.maxBatchDurationMillis = 100000
# Describe the sink
# a1.sinks.k1.type = logger
#a1.sinks.k1.type = file_roll
#a1.sinks.k1.sink.directory = /tmp/flume-log
#a1.sinks.k1.sink.rollInterval = 120
a1.sinks.k1.type=hdfs
a1.sinks.k1.hdfs.path=/flume_log/%y-%m-%d/%H
a1.sinks.k1.hdfs.filePrefix=events-agent
a1.sinks.k1.hdfs.round = true
a1.sinks.k1.hdfs.roundValue = 10
a1.sinks.k1.hdfs.roundUnit = minute
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.fileType = DataStream
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
[/code]
この情報を、Flumeサービスの「設定」から登録します。その前に、保存先のHDFSにディレクトリを作っておきましょう。今回は設定ファイルに /flume_log と指定したので、以下のように行います。

HDFSのディレクトリの準備

[code]
[cloudera@quickstart ~]$ sudo -u hdfs hdfs dfs -mkdir /flume_log
[cloudera@quickstart ~]$ sudo -u hdfs hdfs dfs -chmod 777 /flume_log
[/code]

Flumeを設定する

Flumeサービスの設定には、デフォルトでtier1というAgent名がありました。左側の設定からtier1と入力するか、agentと入力して設定ファイルを表示します。（入力しなくてもいいんですが、見づらいので今回はフィルタ）
このAgent名と設定ファイルを変更します。設定ファイルは上記の設定を貼り付けました。
「変更の保存」ボタンを押して保存します。右上に次のような表示があり、設定が反映されていないことが通知されます。
ボタンを押して反映しましょう。
すると、以下のように差分が表示されます。（注：キャプチャした内容が古いのと、キャプチャーを取るため、意図的にアクセストークンなどは入れていません）

Flumeサービスを起動する

確認して問題がなければ必要に応じてサービスを開始、または再起動しましょう。
通常は上記の手順で設定を保存するとサービスを再起動してくれるはずですが、起動していない場合、または手作業で行う場合は下記のようにCloudera Managerのトップ画面から行います。
これでFlumeサービスが開始し、ログの収集が始まっています。flume-ng agentコマンドの実行は不要です。

Hueでツイートを保存したディレクトリを確認する

Hueのファイルブラウザから /flume_log ディレクトリを見てみましょう。ファイルの保存先を

/flume_log/%y-%m-%d/%H

のようにしたので、/flume_log/年月日/時のようなフォーマットで保存されます。

avro-toolsによるログの閲覧

HDFSに作成されたファイルをダウンロードし、avro-toolsで読んでみます。
[code]
[cloudera@quickstart ~]$ hdfs dfs -ls /flume_log/15-05-26/21Found 2 items
-rw-r–r– 1 flume supergroup 1502688 2015-05-26 21:35 /flume_log/15-05-26/21/events-agent.1432701324514
-rw-r–r– 1 flume supergroup 1536305 2015-05-26 21:37 /flume_log/15-05-26/21/events-agent.1432701423860
[cloudera@quickstart ~]$ hdfs dfs -get /flume_log/15-05-26/21/events-agent.1432701324514
[cloudera@quickstart ~]$ avro-tools tojson –pretty events-agent.1432701324514 |head
log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
{
"id" : "603418779469488128",
"user_friends_count" : {
"int" : 375
},
"user_location" : {
"string" : ""
},
"user_description" : {
"string" : "•God first of all• Hablemos de los Playoffs• Mi Best Friend es pautoso @YovetAlexis•"
（略）
[/code]