q-v-p

vagrant packageのvmware_fusionプロバイダ対応はよ

2014-10-05T18:10:00.002+09:00

調査期間Gの啓示に従いこの辺の記事を読んでもよくわからなかった：

のでおとなしく本家サイトのドキュメントを見に行ったらそもそもpackageコマンドはVirtualBoxしかプロバイダとして想定してないとかふざけたことが書いてあった。VMware Fusionプロバイダのユーザはプラグインのライセンス料支払ってんのに無償プラグイン使ってるユーザより待遇が悪いってどういうことなのHashiCorp.

仕方がないのでこのページ Zsh Script to Create VMware Fusion 5 Vagrant Box を眺めつつ早い話しが

.vmxf, .nvram, .vmsd, .vmx, .vmdk
{"provider":"vmware_fusion"} と記述された metadata.json

が .box っていう拡張子を持ったtar玉にまとめられてればいいということを知ったので手作業でさっさと片付けるなどした。Linuxで使っているときによくお世話になるVMware関連のコマンド群はMacOSXでは /Applications/VMWare Fusion.app/Contents/Library/ の下に転がっているので vmware-vdiskmanager を使ってデフラグ、シュリンクしてあげればOK. できたtar玉は普通に vagrant box add で登録できる。

Clipper Cardの罠・終幕

2014-03-21T21:31:00.002+09:00

すっかり確認するのを忘れてましたが10ドル払い戻しされてました。払い戻しされる1週間くらい前にそれまでやり取りしてたところとは違う部署の担当者から「お前のカードに1ドルキャッシュバックすっから後で確認しろや」的なメールが来てて何で10分の1になってんねん！って突っ込みのメールを送るとかしてたんですが、先方の単純なミスだったようで最終的にはちゃんと戻ってまいりました。

今度は西海岸に行く2週間前くらいにValue Addするようにしよう。これと比べると日本のSuicaとかは遥か先を行ってるよね。

Cloudera Manager 5でNFSを有効にしてみる

2014-03-21T21:18:00.000+09:00

分散ファイルシステムにアクセスする口としてのNFSはクライアントが分散環境の恩恵を受けることができない（NFSクライアントからのサーバコールが1箇所に集約されるので簡単にボトルネックになる）のでそんなにメリットはないんだけど、HDFSの中身を適当にのぞいて回るくらいの用途ならそれなりに便利ではあるので動作確認がてらCloudera Manager 5でHDFS NFS Gatewayを有効にする手順を確認してみます。

まずサービスの一覧からHDFSを選択します：

HDFSサービス画面からインスタンスのタブを選択します：

ここで "Add" ボタンをクリックしてロールの追加を行ないます：

NFS Gatewayを選択します：

Quickstart VMなのでロールを配置するホストに選択肢はありません：

NFS Gatewayが追加されました：

状態が "Stopped" になっているので nfsgateway にチェックを入れて Actions for Selected から開始してやればOKです。これだけでHDFSの / がまるっとexportされます：

Hueで見えてるのとNFSマウントしてファインダから見てるのが同じだということが確認できますね：

ドラッグ＆ドロップでHDFSにファイルを放り込んでみようと思ったんですがユーザIDの対応関係が適当だったんで書き込めませんでした。これはセキュリティ的に正しい姿ですね。

別のLinuxホストにHDFSに書き込む権限を持つユーザと同じユーザID, グループIDを持つユーザを作成してNFS Gateway経由でアクセスしたら問題なく書き込むことができました。

今回はCloudera Manager 5を利用して取り敢えず試してみただけのお知らせでした。@kernel023のブログ記事 CDH5ベータ1のNFSv3ゲートウェイを試してみましたと比較してみるとコマンドラインを一切書いてないのに気付くと思います。コマンドプロンプトとかシェルが嫌いな人には絶賛Cloudera Managerをお勧めします。そうでない人も使うべきというか、これなしで大規模クラスタの運用管理ってムリですよね実際...

続・Clipper Cardの罠

2014-02-16T05:07:00.001+09:00

続き。ここしばらく西海岸行く予定ないんだけど、って連絡したらサポートの中の人から2つのオプションが提示された：

次に西海岸に行く機会があったら1週間前くらいに連絡くれればカードに取り込めるようにしてあげる

いまアカウントに設定されているカード情報が支払いに使ったのと同じカードだったら払い戻ししてあげる

ここは迷わず払い戻しを選択。ただし手続きして承認されて払い戻しされるまでに30日くらい掛かるよ、って念を押された。戻ってくるなら何でもいいので一月くらい待ちますわ。そっちこそリファンド、忘れんなよw

Clipper Cardの罠

2014-02-13T00:31:00.000+09:00

一部ですこぶる便利だと評判のClipper Cardですが割としょーもない落とし穴があるので自分のために記録しておきます。

ウェブサイトでチャージしてもシステムに反映されるまでに3〜5日を要する

システムに反映された金額をカードに取り込むにはカードリーダにかざさなきゃダメ

ウェブサイトでチャージしたおカネは半年以内にカードに取り込まないと失効する

去年 San Francisco 〜 Palo Alto 〜 San Jose 辺りをうろうろしていたときにすこぶる便利に使っていたんだけど、そろそろ残高心もとなくなってきたからチャージしとこうと思ってチャージしたのが帰国の2日くらい前。そのまま帰国して経費精算するのに乗車履歴見てたらチャージした筈の金額が取り込まれてなくて、システムに反映されるまでにタイムラグがあるのを知ってまぁいいかと思って放って置いて、先月 San Francisco 行ったときに使ったら（=カードリーダにかざしたら）チャージされると思っていたのに実は失効していたという罠。

履歴調べて失効した分を再度カードに取り込めるようにしてくれる、ってサポートから連絡きたけどここ半年以内に西海岸行く予定ないんだよなぁ... どーすべw

Java 7 Update 51 on MacOSX 10.9.1

2014-01-18T20:10:00.000+09:00

Java7 Update 51が来てたので早速インストールしておいたんだけど：

shellから確認するとJava 7 Update 45のままなのはどうしてなの？

2つのTwitterSource.java

2014-01-13T16:30:00.000+09:00

CDH 4.5に含まれるFlume-NGにはTwitterからデータを取得するためのTwitterSourceクラスが含まれるようになっています。これがCDH 4.5にTwitter4Jが同梱されるようになった理由。実際こんな感じでjarファイルが転がっているのが見て取れます。

% cd /opt/cloudera/parcels/CDH-4.5.0-1.cdh4.5.0.p0.30/lib/flume-ng/lib
% ls -lF twitter*
-rwxr-xr-x 1 root root 283752 Nov 21 11:07 twitter4j-core-3.0.3.jar*
-rwxr-xr-x 1 root root  27690 Nov 21 11:07 twitter4j-media-support-3.0.3.jar*
-rwxr-xr-x 1 root root  56262 Nov 21 11:07 twitter4j-stream-3.0.3.jar*

これだけで終わってもいいのですが折角なので混乱を招くのではないかと思うTwitterSource.javaについて書いておきます。

世の中には2つのTwitterSource.javaがある

https://github.com/cloudera/cdh-twitter-example で提供されるもの
Flume-NGの機能として提供されるもの

CDH 4.5に入っているのはこれ。CDH 5にも入るハズ

cdh-twitter-example版

com.cloudera.flume.source.TwitterSource
TwitterのSample APIを叩く

指定されたキーワードを含むものが取得できる

フォーマットはいじらずに後段に流す

コードからコメント抜いてくるとこう書いてある：

/**
 * A Flume Source, which pulls data from Twitter's streaming API. Currently,
 * this only supports pulling from the sample API, and only gets new status
 * updates.
 */

Flume-NG版

org.apache.flume.source.twitter.TwitterSource
TwitterのStreaming APIを叩く

Firehoseで取得できるものの1%がサンプリングされるらしい

Avroフォーマットにして後段に流す

コードからコメント抜いてくるとこう書いてある：

/**
 * Demo Flume source that connects via Streaming API to the 1% sample twitter
 * firehose, continously downloads tweets, converts them to Avro format and
 * sends Avro events to a downstream Flume sink.
 *
 * Requires the consumer and access tokens and secrets of a Twitter developer
 * account
 */

`まとめ`

早い話がClouderaが作ったTwitterSource便利だからApache Software Foundationに寄贈してFlume-NGに取り込まれたけどちょっと機能が変わってますね、ということでした。

CDH 4.5に移行したらFlume agentが動かなくなった

2014-01-12T02:07:00.001+09:00

年末年始の休みでアップグレードしようと思ってたのが作業できてなかったので、忘れないうちにと思って3連休の初日にCloudera Manager 4.8 + CDH 4.5の環境に移行してみた。移行作業自体は何の問題も無く終わったんだけど新しくしてから全サービス起動して問題ないか見て回ってたら不審な点に気付いてしまった：

Flumeからデータが流れてきてないw

いやこれ本番環境とかだったら全然笑えない事態だけど。Cloudera Managerからログを調べてみたらagentがエラー吐いてた：

Unable to start EventDrivenSourceRunner: { source:com.cloudera.flume.source.TwitterSource{name:Twitter,state:IDLE} } - Exception follows.
java.lang.NoSuchMethodError: twitter4j.FilterQuery.setIncludeEntities(Z)Ltwitter4j/FilterQuery;
 at com.cloudera.flume.source.TwitterSource.start(TwitterSource.java:141)
 at org.apache.flume.source.EventDrivenSourceRunner.start(EventDrivenSourceRunner.java:44)
 at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)
 at java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)
 at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)
 at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)
 at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
 at java.lang.Thread.run(Thread.java:662)

Twitter4Jのところで存在しないメソッドを呼び出すとかアホなことしてる。これ原因は判ってしまえば簡単で、CDH 4.5に入っているTwitter4JのバージョンとFlume agentが使っているTwitter4Jのバージョンが違うのが問題だった。以前のCDHにはTwitter4Jは含まれてなかったのでFlume agentのjarに入れてあったんだけどCDH 4.5側のTwitter4Jを先に見付けてしまって誤動作するという昔懐かしいDLL HELLみたいな状況が起きてた。

判ってしまえば対処も簡単でCDH 4.5用にpomを書き換えてFlume agentをビルドし直してデプロイ、サービス再起動で問題なくデータが流れ始めた。

fork元の cloudera/cdh-twitter-example に差分が来てたのでそれらを kmizumar/cdh-twitter-example に取り込んだだけ。

どうしてCDH 4.5にTwitter4Jが入っているのか、は気が向いたら書きます。

cdh-twitter-exampleで遊ぶ

2013-12-10T00:59:00.000+09:00

この記事はHadoop Advent Calendar 2013, 13日目のエントリです。ブログなんざ書いたことないので勝手が良く分かりませんが@sudabonさん@kernel023さんのお前も何か書けやコラという圧力に負けてBloggerにページを用意してみました。

Hadoopの利用目的としてデータの分析に取り組んでいる、もしくはこれから取り組もうとしているユーザは多いのではないかと思います。中でもログ分析と並んでソーシャルメディアの分析は各社が積極的に取り組もうとしているテーマであり、私が勝手にTwitter分析3部作と呼んでいる以下のブログを読まれた方も多いのではないでしょうか：

まだ読まれていない方のためにざっくりと説明しますと、Flumeを利用してTwitterからデータをぶっこ抜いてきてHDFSに流し込んでやれば、あとはHiveにJSON SerDeを組み込むだけでSQLでTwitterを分析する準備ができちゃいます、ということが紹介されてます。

前置きが長くなりましたがこのエントリでは上の図のTwitter分析用のデータパイプラインに細工してデータを流し込む先としてHBaseを追加してみたり、HBaseに書き込まれたデータについてインデックスを作成して全文検索が出来るようにする方法を紹介してみたいと思います。せっかちな方のためにどんな技術トピックが関係するかを先に列記しておくとFlume Interceptor, Flume flow multiplexer, Flume HBaseSink, Lily HBase NRT Indexer Service, Cloudera Search (SolrCloud)といったものになります。ここから先はTwitter分析3部作のブログは読んでいる、もしくは同等の内容を知っているものとして話を進めます。Twitter分析3部作のオリジナルのコードはGithubの以下のリポジトリに置かれています：

https://github.com/cloudera/cdh-twitter-example

これを勝手にforkしていじくったものをGithubの以下のリポジトリに置いています。興味がある人はのぞいてみてください。

https://github.com/kmizumar/cdh-twitter-example

InterceptorでFlume eventをネコババする

先に見たデータパイプラインにある「カスタム作成のFlume Source」の正体はcom.cloudera.flume.source.TwitterSourceクラスです。このクラスは設定ファイルで指定されたキーワードで検索したtweetをJSONフォーマットで出力してきます。1イベントについて1つのJSONレコードが含まれています。以下は例です：

{"filter_level":"medium","contributors":null,"text":"In the past, our team at Bing Ads has applied big data insight to ad copy strategy in the travel and financial servi\u2026http://t.co/A3A5xRJE8b","geo":null,"retweeted":false,"in_reply_to_screen_name":null,"possibly_sensitive":false,"truncated":false,"lang":"en","entities":{"symbols":[],"urls":[{"expanded_url":"http://lnkd.in/bipZAX3","indices":[117,139],"display_url":"lnkd.in/bipZAX3","url":"http://t.co/A3A5xRJE8b"}],"hashtags":[],"user_mentions":[]},"in_reply_to_status_id_str":null,"id":390880848465375232,"source":"<a href=\"http://www.linkedin.com/\" rel=\"nofollow\">LinkedIn<\/a>","in_reply_to_user_id_str":null,"favorited":false,"in_reply_to_status_id":null,"retweet_count":0,"created_at":"Thu Oct 17 16:43:51 +0000 2013","in_reply_to_user_id":null,"favorite_count":0,"id_str":"390880848465375232","place":null,"user":{"location":"Raleigh, NC","default_profile":true,"statuses_count":3010,"profile_background_tile":false,"lang":"en","profile_link_color":"0084B4","id":474950294,"following":null,"favourites_count":0,"protected":false,"profile_text_color":"333333","description":null,"verified":false,"contributors_enabled":false,"profile_sidebar_border_color":"C0DEED","name":"Dan Damato","profile_background_color":"C0DEED","created_at":"Thu Jan 26 14:35:17 +0000 2012","default_profile_image":false,"followers_count":139,"profile_image_url_https":"https://si0.twimg.com/profile_images/1785237335/images_normal.jpg","geo_enabled":false,"profile_background_image_url":"http://abs.twimg.com/images/themes/theme1/bg.png","profile_background_image_url_https":"https://abs.twimg.com/images/themes/theme1/bg.png","follow_request_sent":null,"url":null,"utc_offset":null,"time_zone":null,"notifications":null,"profile_use_background_image":true,"friends_count":389,"profile_sidebar_fill_color":"DDEEF6","screen_name":"onlinesalesnut","id_str":"474950294","profile_image_url":"http://a0.twimg.com/profile_images/1785237335/images_normal.jpg","listed_count":1,"is_translator":false},"coordinates":null}

JSONレコード1つをHBaseの1カラムに格納してもいいのですが、ここではtweetの中から取り敢えず興味のある

id : tweetを一意に識別する識別子（だと予想）
text : tweetの本文
user.screen_name : twitterユーザのアカウント（例：kmizumar）
user.name : twitterユーザのフルネーム（例：Kiyoshi Mizumaru）

だけを抽出してそれぞれHBaseのtid, text, screen, userというカラムに格納することにします。これを実現するためにはJSONレコードをパースして興味のある名前と値のペアだけを抜き出すという処理が必要になりますが、このためにTwitterSourceクラスを改造することは得策ではありません。Flumeには任意のSourceにくっつけてそのSourceへ入力されてくるイベントを横取りして好きなようにイベントに細工を施すことのできるInterceptorが用意されてますのでこれを使います。TwitterSourceにくっつけて上記の処理を行わせるために作成したものがcom.cloudera.flume.interceptor.TwitterInterceptorクラスです：

https://github.com/kmizumar/cdh-twitter-example/blob/master/flume-sources/src/main/java/com/cloudera/flume/interceptor/TwitterInterceptor.java

private byte[] readJsonStream(InputStream)メソッド

JSONレコードをパースして”id”, “text”, “user.name”, “user.screen_name”に該当する値だけを抜き出します。抜き出した値を使用して

____ID____:<idの値>____TEXT____:<textの値>____USER____:<user.nameの値>____SCREEN____:<user.screen_nameの値>

という内容を持つbyte[]を組み立てて返します。パーサーはcom.google.gson.stream.JsonReaderクラスを利用しています。

pubic Event intercept(Event)メソッド

イベントをひとつ受け取ったら

元のイベントのヘッダをコピーしたものに SINKTYPE → HBASE というデータを追加
上記ヘッダとreadJsonStreamから戻ってきた内容をボディに持つ新しいイベントを作成

して返します。ここではイベントのボディを完全に別の物にすり替えています。

public List<Event> intercept(List<Event>)メソッド

イベントのリストを受け取ります。リストに含まれているイベントそれぞれについて

元のイベントのヘッダに SINKTYPE → HDFS というデータを追加する
Event intercept(Event)メソッドを呼び出して SINKTYPE → HBASE の新しいイベントを取得する

という処理を行いHDFS用とHBase用の2つのイベントを作成してリストに入れて返します。入力されてきたイベントのリストはHDFS用とHBase用のイベントを含む倍の長さのイベントのリストになって出力されることになります。

仕方がないのでTwitterSourceを改造する

ここまでで作成したTwitterInterceptorを使うことで、TwitterSourceに流れてくるイベントを横取りしてHDFS向けに従来通りのJSONレコードを保持するevent、HBASE向けにJSONレコードから欲しいところだけ抜き出した内容を保持するevent、の2つのeventを吐き出すことができるようになるはずです。はずです、と書いたのには理由があって実はオリジナルのTwitterSourceクラスの実装では期待した結果が得られません。これはTwitterSourceクラスの中で使用しているStatusListenerクラスのインスタンスがJSONレコードをラップするイベントオブジェクトを作成したら、イベントオブジェクト1つを引数にして後段のチャネルに流しているのが原因です。これではTwitterInterceptorクラスのEvent intercept(Event)メソッドしか呼ばれないためイベントを捨てるか（nullを返す）、イベントを改変して返すか、そのまま返すか、しかできることがありません。現在のTwitterInterceptorの実装ではすべてのイベントがHBase用に置き換えられて出てくるだけで、いずれにしてもこのままでは今回やろうとしている「1つのイベントを受け取って2つのイベントを返す」ということはできないのでチャネルに流す前にイベントオブジェクトをList<Event>に突っ込んでチャネルにはリストを流すように変更してしまいます。もっと上手い方法はありそうですがQuick and Dirty Hackでここは乗り切ることにします。

https://github.com/kmizumar/cdh-twitter-example/commit/840fb2279b11437d850027b5feeb880b215cb2a9#diff-c3d9e77e0883358123b3a4fb832b2d87


diff --git a/flume-sources/src/main/java/com/cloudera/flume/source/TwitterSource.java b/flume-sources/src/main/java/com/
index 53c2f60..0305702 100644
--- a/flume-sources/src/main/java/com/cloudera/flume/source/TwitterSource.java
+++ b/flume-sources/src/main/java/com/cloudera/flume/source/TwitterSource.java
@@ -18,7 +18,9 @@
 
 package com.cloudera.flume.source;
 
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.flume.Context;
@@ -110,8 +112,8 @@ public class TwitterSource extends AbstractSource
         headers.put("timestamp", String.valueOf(status.getCreatedAt().getTime()));
         Event event = EventBuilder.withBody(
             DataObjectFactory.getRawJSON(status).getBytes(), headers);
-        
-        channel.processEvent(event);
+       List<event> events = Arrays.asList(event);
+        channel.processEventBatch(events);
       }
       
       // This listener will ignore everything except for new tweets

Flow multiplexerでFlume eventを撃ち分ける

Flumeのデータフローを組み立てます。登場するのはsourceとしてTwitterからデータを入力するものが1つ、sinkとしてHDFSへ流し込むのとHBaseへ流し込むものが2つ、そしてそれぞれのsinkに対応するchannelが2つです。flume.confからポイントとなる部分だけ抜粋します。TwitterAgentという名前でエージェントを作成しています。

TwitterAgent.sources = Twitter
TwitterAgent.channels = MemChannel MemChannel2
TwitterAgent.sinks = HDFS HBASE

TwitterAgent.sources.Twitter.type = com.cloudera.flume.source.TwitterSource
TwitterAgent.sources.Twitter.channels = MemChannel MemChannel2

続いてこのTwitterAgentにインターセプターを設定します。インターセプターの型を指定する*.typeプロパティにはインターセプターのクラス名ではなくインターセプターのビルダークラスを指定しなければならないのに気付くまでかなり時間をムダに費やしたのは秘密です。

TwitterAgent.sources.Twitter.interceptors = TwitterInterceptor
TwitterAgent.sources.Twitter.interceptors.TwitterInterceptor.type = com.cloudera.flume.interceptor.TwitterInterceptor$Builder

これでTwitterSourceが取得してきたイベントはTwitterInterceptorを経由してからチャネルに流れ出るようになります。どのイベントをどのチャネルに流し込むかを選択するにはsourceにマルチプレクサを設定して、イベントのどのヘッダを見て、その内容に応じてどのチャネルに流し込んで欲しいかを指定します。ここではSINKTYPEヘッダの内容を見てそれが"HDFS"だったらMemChannelへ、"HBASE"だったらMemChannel2へ流し込むようにしています。いまのTwitterInterceptorの実装ではSINKTYPEがHDFSかHBASEのどちらかしか出力してきませんが、将来的に他のタイプのものを出力するようにしたときにHBase側に流れても正しく扱えないのでデフォルトをHDFS側のMemChannelに倒しています（HDFSなら任意のJSONレコードをそのまま格納できるので）。

TwitterAgent.sources.Twitter.selector.type = multiplexing
TwitterAgent.sources.Twitter.selector.header = SINKTYPE
TwitterAgent.sources.Twitter.selector.mapping.HDFS = MemChannel
TwitterAgent.sources.Twitter.selector.mapping.HBASE = MemChannel2
TwitterAgent.sources.Twitter.selector.default = MemChannel

HBaseSinkでHBaseにデータを流し込む

MemChannel側の設定はTwitter分析3部作のブログと変わらずHdfsSinkに接続してHDFSに書き込んでいるだけなので説明は省略します。MemChannel2側のチャネルにはHBaseSinkを接続してデータをHBaseに流し込むことにします。

TwitterAgent.sinks.HBASE.channel = MemChannel2
TwitterAgent.sinks.HBASE.type = hbase
TwitterAgent.sinks.HBASE.table = tweets_hbase
TwitterAgent.sinks.HBASE.columnFamily = cf1
TwitterAgent.sinks.HBASE.batchSize = 1000
TwitterAgent.sinks.HBASE.serializer = org.apache.flume.sink.hbase.RegexHbaseEventSerializer
TwitterAgent.sinks.HBASE.serializer.regex = ^____ID____:(.*)____TEXT____:(.*)____USER____:(.*)____SCREEN____:(.*)$
TwitterAgent.sinks.HBASE.serializer.colNames = tid,text,user,screen

ここでは次のことを指定しています：

書き込む先のHBaseテーブルの名前をtweets_hbase
カラムファミリ名をcf1
RegexHbaseEventSerializerを使ってイベントボディからデータを取り出す（ここ正しくはデシリアライザじゃないのかなぁ... 名前）
データを取り出すための正規表現パターン
マッチしたデータをHBaseテーブルのどのカラムに書き込むか

ここまでの内容に合わせてHBaseにテーブルを作成しておきます。

% hbase shell
hbase(main):001:0> create 'tweets_hbase', 'cf1'

ここまでの構成が問題なくできていればFlume agentを立ち上げることでTwitterからデータの取り込みが開始されてHDFS, HBaseの双方にデータが書き込まれて行きます。画面はHueのHBaseブラウザでtweets_hbaseテーブルを表示してみたところです。指定したcf1:screen, cf1:tid, cf1:text, cf1:userカラムにそれぞれ値が格納されている様子がわかります：

Lily HBase NRT Indexer Serviceでインデックス作成

そろそろ飽きてきたころだと思いますがここまでくればもう少しです。まずはインデックスを作成したいHBaseのテーブルでレプリケーションが有効になっていることが要求されますのでこれを片付けておきます：

% hbase shell
hbase(main):001:0> disable 'tweets_hbase'
hbase(main):002:0> alter 'tweets_hbase', {NAME => 'cf1', REPLICATION_SCOPE => 1}
hbase(main):003:0> enable 'tweets_hbase'

立て続けにmorphlines.confを設定します。後ほど作成するSolrコレクション　hbase-collection1に対してインデックスを登録していくこと、HBaseテーブルのどのカラムを持ってきてSolrスキーマのどのフィールドに割り当てるか、を指定しています。この例ではcf1:textカラムの値がSolrスキーマのtextフィールドとして出力されることになります。


SOLR_LOCATOR : {
  # Name of solr collection
  collection : hbase-collection1

  # ZooKeeper ensemble
  zkHost : "$ZK_HOST" 
}


morphlines : [
  {
    id : morphline
    importCommands : ["com.cloudera.**", "com.ngdata.**"]

    commands : [                    
      {
        extractHBaseCells {
          mappings : [
            {
              inputColumn : "cf1:tid"
              outputField : "tid" 
              type : string 
              source : value
            }
            {
              inputColumn : "cf1:text"
              outputField : "text" 
              type : string 
              source : value
            }
            {
              inputColumn : "cf1:user"
              outputField : "user" 
              type : string 
              source : value
            }
            {
              inputColumn : "cf1:screen"
              outputField : "screen" 
              type : string 
              source : value
            }
          ]
        }
      }

      { logDebug { format : "output record: {}", args : ["@{}"] } }
    ]
  }
]

Solrのコレクションを作ります。スキーマ定義は上記のmorphlines.confに合わせて書き換えたものをGithubに置いてますのでこれを利用してSolrCloudにインスタンスディレクトリの作成とコレクションの作成を行います。下の例ではコレクションの作成時にshardの数を4と指定していますが利用しているSolrCloudのノード数に合わせて適宜変更する必要があるので注意してください。Solrスキーマの詳細について知りたいときはWebサイトや専門の書籍を参照してください。深遠なる検索システムの世界が貴方を待っています。

% solrctl instancedir --create hbase-collection1 hbase-collection1
% solrctl collection --create hbase-collection1 -s 4

最後にインデクサの設定ファイルを作成します。この設定ファイルではインデクサがどのHBaseテーブルを読みに行って、どのmorphlines.confの設定に従って動作するべきなのかを指定します。Githubにmorphline-hbase-mapper.xmlという名前で置いてあるサンプルの内容を以下に示します。ここでtweets_hbaseテーブルをインデックスの作成対象として指示しています。パラメータmorphlineFileの値がフルパスで指定されていないのはCloudera Managerの管理下にあるSolrCloudを利用しているときの設定の仕方になります。

<?xml version="1.0"?>
<indexer table="tweets_hbase" mapper="com.ngdata.hbaseindexer.morphline.MorphlineResultToSolrMapper">
   <!-- The relative or absolute path on the local file system to the morphline configuration file. -->
   <!-- Use relative path "morphlines.conf" for morphlines managed by Cloudera Manager -->
   <param name="morphlineFile" value="morphlines.conf"/>
   <!--
   <param name="morphlineFile" value="/etc/hbase-solr/conf/morphlines.conf"/>
   -->
   <!-- The optional morphlineId identifies a morphline if there are multiple morphlines in morphlines.conf -->
   <!-- <param name="morphlineId" value="morphline1"/> -->
</indexer>

準備がすべて整ったところでインデクサを登録します。動作しているLily HBase NRT Indexer Serviceに対して設定ファイルの登録を行い、正しく登録されていることを確認できたらすべての作業は終了です。登録に用いるhbase-indexer add-indexerコマンドの実行例を以下に示します。ZooKeeperアンサンブルの指定は利用している環境に合わせて適宜変更してください。

% hbase-indexer add-indexer -n myindexer \
  --indexer-conf morphline-hbase-mapper.xml \
  -cp solr.zk=nn1.demo.dev,nn2.demo.dev,ms3.demo.dev/solr \
  -cp solr.collection=hbase-collection1 -z nn1.demo.dev

インデクサの一覧を見るにはhbase-indexer list-indexersコマンドを使用します。設定ファイルに間違いがあるときは一番下に表示されるProcessesの項目に実行中のプロセスが1個もなくてshard数分のプロセスがfailしているという状態になりますので注意して見てください：

% hbase-indexer list-indexers -z nn1.demo.dev
Number of indexes: 1

myindexer
  + Lifecycle state: ACTIVE
  + Incremental indexing state: SUBSCRIBE_AND_CONSUME
  + Batch indexing state: INACTIVE
  + SEP subscription ID: Indexer_myindexer
  + SEP subscription timestamp: 2013-12-03T03:38:40.206+09:00
  + Connection type: solr
  + Connection params:
    + solr.collection = hbase-collection1
    + solr.zk = nn1.demo.dev,nn2.demo.dev,ms3.demo.dev/solr
  + Indexer config:
      645 bytes, use -dump to see content
  + Batch index config:
      (none)
  + Default batch index config:
      (none)
  + Processes
    + 4 running processes
    + 0 failed processes

CDH4.4 + Cloudera Manager 4.7の組み合わせではCloudera SearchとCloudera Managerの統合がそんなに進んでいないのでSolrCloudの構成状況やドキュメントの登録状況を確認するためにはSolr Admin Console画面を参照します。CollectionのStatistics画面で時間が経つにつれて登録されているドキュメントの数（= tweets_hbaseテーブルに格納されたtweetの数）がどんどん増えて行く様子が見て取れます。またHueのSearch AppでCollection Managerからhbase-collection1を登録して適当なキーワードを投げたりして遊んでみてください。

おわりに

駆け足でTwitterSourceからデータを横取りしてHBaseに放り込み全文検索用のインデックスを張るところまでを眺めてみました。Flumeからしかデータを書き込まないのであればFlume Morphline Solr Sinkを利用してインデックス作成とデータ書き込みを同時にやってのけるという芸当も実際には可能です。興味のある人は是非挑戦してみてください。

Hadoop Advent Calendar 2013, 14日目はs-wool@githubさんです。おたのしみに。

投稿テスト

2013-12-08T00:51:00.001+09:00