HBase0.89におけるトランザクションについて

[sincere08]

皆様

初めまして。 sincere08 と申します。

表題の件で質問なのですが、HBase0.89でトランザクション処理の実現方法が分からずにいます。
0.20系から0.89に移行した時にHBaseにcontribが含まれなくなったということで、0.89以降はGitHubからトランザクション関
係のソースを持ってくる認識でいます。

試そうとしている条件は
テーブル複数（以下の例ではA,B,Cの3つ）にデータをputする際に
例えば、A、Bにputした後にCでのputに失敗
　→A、Bに入れたデータをなかったことにしたい。
（put失敗後に手動でのA,Bのdeleteを行うのは無し）

0.20系の場合、内包されている『hbase-0.20.x-transactional.jar』をそのまま使用
0.90の場合、『http://wiki.apache.org/hadoop/SupportingProjects』のHBase-trxを使
用
どちらのパターンも動作確認は取れたのですが、0.89のみ実施ができていません。

HBase-trxのREADME.txtを読む限りだとに0.89のトランザクションのsrc（もしくはlib）が
どこかにありそうなのですが、Branch Listやその他を探してみても見つけることができません。

0.89に拘る理由はCloudera（CDH3）にあるHadoop及びHBaseを使用したいためです。
※2010年11月30日現在、HBase0.89のみサポートされているようです。

0.20系、0.90のトランザクションのjarを0.89で使用するとコンパイルエラーになるため、
最悪、自分で0.90のsrcを0.89で使用可能なように書き換えるしかないのかと考えています。
ざっと見た感じ、THLogやResionServerやleaseの修正だと思うのですが、やはりデグレの心配もあります。


前段が長くなりましたが、質問としては
１．GitHub、HBase-trxでHBase0.89のトランザクションのソースの在り処をご存知の方がいますでしょうか。
２．GitHub、HBase-trxに頼らず、自前でトランザクションを行う良い方法をお持ちの方がいますでしょうか。
３．GitHub、HBase-trxのHBase0.90のソースをHBase0.89で使用できるように修正したことがある方、もしくは修正方法の
検討がついている方がいますでしょうか。

HadoopというよりHBaseの質問になってしまい、すみません。
答えられる範囲で構いませんので、何か質問や意見等がありましたら是非お返事いただければと思います。

以上、よろしくお願いします。

[Tatsuya Kawano]

sincere08さん、こんにちは。

HBase 0.89 ですが、これは、開発途中の 0.90 の定期的なスナップショットですので、本番では使用しないほうがいいですよ。0.89は、私たちHBaseの利用者がお試しして、0.90の開発にフィードバックできるように用意されたものです。ですから、品質よりも、定期的にリリースすることが重視されています。別の言い方をすると、0.89は、0.90のアルファ版、ベータ版です。

また、Clouderaの CDH3も、現在のところ、フィードバックを目的としたベータ版（b3）です。

> １．GitHub、HBase-trxでHBase0.89のトランザクションのソースの在り処をご存知の方がいますでしょうか。

ブランチのネットワークを見ると、

https://github.com/hbase-trx/hbase-transactional-tableindexed/network

9/29に HBase 0.89でビルドできたブランチ（黄緑色の線）は、masterにマージされて消滅し、現在は 0.90 RC対応版のみが残ってますね。そもそも、0.89は、開発途中の0.90の、ある日のスナップショットですので、0.89をターゲットに開発、ということはしていないと思います。

> ２．GitHub、HBase-trxに頼らず、自前でトランザクションを行う良い方法をお持ちの方がいますでしょうか。

HBase-trxがHBaseに取り込まれないのは、それがなくても、HBase流にテーブル設計することで目的が果たせるからです。A、B、Cのテーブルの関係をもう少し詳しく教えてもらえないですか？

たとえば、AとBは、マスターとディテールの関係でしょうか？（A～Bが１対多。注文と注文明細のような関係。Bに単独でアクセスすることはなくて、Aに最初にアクセスして、そこからリレーションをたどって、Bを見つけるパターン）

あるいは、Bは、Aに対する二次インデックス的なものですか？（A～Bが多対１ or １対１。Aに直接アクセスすることもあれば、まずBにアクセスして、そこからリレーションを辿って、Aにたどり着くパターン）

またあるいは、AとBはお互いにリレーションを持たないけれども、内容に一貫性を持たせる必要があるのでしょうか？（データと、データに対する操作の監査ログ）

何か例を挙げてもらえるとアドバイスしやすいです。

よろしくお願いします。

たつや

--
Tatsuya Kawano (Mr.)
Tokyo, Japan

http://twitter.com/#!/tatsuya6502

[sincere08]

たつやさん

ご返信ありがとうございます。

0.89はベータ版であり本番での使用には注意深くある必要があるという旨、了解いたしました。

＞またあるいは、AとBはお互いにリレーションを持たないけれども、内容に一貫性を持たせる必要があるのでしょうか？（データと、データに対する操作の
監査ログ）
テーブルの-関係としてはこちらの例に近いです。
RDBで言うところの外部キーの役割は持っていません。

抽象的な書き方で申し訳ありませんが、
Aだけ持っていてB,Cは持っていないというのは避けたい all or nothing の考えです。
それぞれのテーブルはfamilyは一つずつですが、keyやqualifierは異なり、valueは3テーブルとも同じ。
また、1つのkeyに複数のvalueがある場合もあります。
様々な検索条件に対応できるように複数のテーブルに同じデータを持たせています。

＞テーブル複数（以下の例ではA,B,Cの3つ）にデータをputする際に

＞例えば、A、Bにputした後にCでのputに失敗
＞　→A、Bに入れたデータをなかったことにしたい。

先の例では、上記のように書きましたが、実際は失敗に限らないです。
A操作後（table.put()等した後）にシステム上のとある条件で登録したデータを
戻したいといった場合にどのように対処するのがベストなのか確認したかったです。
何かよい案はありますでしょうか？

HBase流のテーブル設計ですか。
1テーブルにデータをまとめてしまって
keyやqualifierにテーブル名＋値　等も考えたのですが、
テーブル数や名前が可変のため、また値の種類も膨大のため
filterやscanを使用しての検索処理が複雑になるという事情で分けてしまった経緯があります。
一貫性を求めるシステムで今の構造の場合、やはり再設計を検討したほうがよいのでしょうかね。

以上です。

[Tatsuya Kawano]

sincere08さん、

たつやです。こんばんは。

>> またあるいは、AとBはお互いにリレーションを持たないけれども、内容に一貫性を持たせる必要があるのでしょうか？（データと、データに対する操作の監査ログ）
>>

> テーブルの-関係としてはこちらの例に近いです。
> RDBで言うところの外部キーの役割は持っていません。

> それぞれのテーブルはfamilyは一つずつですが、keyやqualifierは異なり、valueは3テーブルとも同じ。
> また、1つのkeyに複数のvalueがある場合もあります。
> 様々な検索条件に対応できるように複数のテーブルに同じデータを持たせています。

では、社員テーブルのようなものを想像することにします。

A が、社員番号を行キーにしたテーブル、１行に社員１名の属性
B が、氏名を行キーにしたテーブル、１行に社員１名〜複数名の属性
C が、部課名を行キーにしたテーブル、１行に社員複数名の属性

A、B、Cそれぞれのテーブルに、社員の属性として、社員番号、氏名、部課名が格納されている。もちろん、こんなに単純ではないと思いますが...。

> Aだけ持っていてB,Cは持っていないというのは避けたい all or nothing の考えです。

人事異動で社員の部課名を更新したり、社員の１人が結婚して姓を更新したりするときに、Aで検索したときと、B or Cで検索したときに矛盾しない結果を得たい。ということですね。

> HBase流のテーブル設計ですか。
> 1テーブルにデータをまとめてしまって
> keyやqualifierにテーブル名＋値　等も考えたのですが、
> テーブル数や名前が可変のため、また値の種類も膨大のため
> filterやscanを使用しての検索処理が複雑になるという事情で分けてしまった経緯があります。

以下の、マスター・ディテール関係の場合は、１テーブルにまとめることを提案しようと思っていました。

>> たとえば、AとBは、マスターとディテールの関係でしょうか？（A～Bが１対多。注文と注文明細のような関係。Bに単独でアクセスすることはなくて、Aに最初にアクセスして、そこからリレーションをたどって、Bを見つけるパターン）

しかし、今回のケースでは、様々な検索条件に対応させたいということですので、１つのテーブルにまとめることは難しいと思います。複数のテーブルがあるということは、sincere08のおっしゃるとおり、更新にはトランザクションが必要となります。

HBase-trxは、HBaseを複数行・複数テーブルの「ACIDトランザクション」に対応させるためのものですが、ここでは、別の選択肢として、Outerthoughtが開発した HBase RowLogライブラリ（クライアント側で動くライブラリ）を使って、「BASEトランザクション」を実現する方法を紹介します。

BASEトランザクションは、Basically Available、Soft state、Eventually consistentの頭文字です。分散システムに適した実行モデルで、A、B、C各テーブルの更新をロックを取得せずに行います。ロックを使いませんので、更新の中間状態が外部から見えるわけで、そこのところがACIDとは異なります。しかし、最後には必ず all or nothingになることを保証します。

BASEトランザクションは、例えると、銀行間の振り込みのような処理でしょうか。銀行Aの口座で、銀行Bの口座への振り込みを実行しても、銀行Bの口座にお金が届くまでにはタイムラグがあります。しかし、最終的には、all or nothing、つまり、銀行Bの口座にお金が届くか、あるいは、銀行Aにお金が戻ってくるかのどちらかの状態になり、一貫性が得られます。


では、HBase と RowLog ライブラリによる BASE トランザクションの実装を紹介します。

RowLogライブラリはここからダウンロードできます。
http://www.lilyproject.org/lily/about/downloads.html

lily-0.2.1.tar.gz と lily-0.2.1-src.tar.gz をダウンロードして下さい。前者に jar が入っていますが、Javadocが入っていないので、後者のソースコードから、Javadocを自分で作ることになります。（Outerthoughtで作ってもらえるよう、お願いしてみます）

ちなみに、mvn -P fast install を実行したところ、HBase 0.89.20100924+28-lily-0.2 がダウンロードされました。HBase 0.90 だとコンパイルできません。


RowLogの解説はこちらです。
http://outerthought.org/blog/449-ot.html
http://www.lilyproject.org/lily/about/playground/hbaserowlog.html


ロジックはこんな感じです。
================================================

// Write Ahead ログ
private RowLog wal;

// BテーブルとCテーブルを非同期で更新するためのバックグラウンドプロセス。
private RowLogProcessor processor;

private isInitialized = false;


/**
* RowLog利用の準備。最初に１回だけ実行。
*/
public initializeWal() {
if (isInitialized) return;

wal = new RowLogImpl(...);

// Aテーブルを wal として使うように設定する。
// TableBCUpdater クラス（後述）を MessageListener として登録する。
// subscriptionを設定する。
//
// 具体的なコードは以下を参照。
// lily-src-0.2.1/global/rowlog/impl/src/test/java/org/lilyproject/rowlog/impl/test/Example.java
...
...
...

// RowLogの非同期プロセッサーを起動しておく
// プロセッサーはクラスター全体で１つのプロセスだけがアクティブになる。
// これにより、walに登録した RowLogMessage が、登録した順序で
// 実行されることを保証する。
//
// RowLogライブラリーでは、ZooKeeperを使って、複数のプロセッサーの
// なかから、アクティブにするものを１つだけ選出する。
RowLogProcessor processor = new RowLogProcessorImpl(rowLog, configurationManager);
processor.start();

isInitialized = true;
}


/**
* A、B、Cテーブルの更新。
*
* @return 更新に成功したら true、失敗したら false
*/
public boolean updateABC( ... ) {

// === ステップ１ ===
// 各テーブルの put の準備

Put putA = new Put(rowA);
putA.set(更新後の値);

Put putB = new Put(rowB);
putB.set(更新後の値);

Put putC = new Put(rowC);
putC.set(更新後の値);


// === ステップ２ ===
// BテーブルとCテーブルの更新内容を RowLog に登録する。
// これらは、Aテーブルの rowA 行の RowLog 用カラムファミリーに登録される。

RowLogMessage walB = wal.putMessage(rowA, Bytes.toBytes("tableB"),
シリアライズしたputB, putA)
RowLogMessage walC = wal.putMessage(rowA, Bytes.toBytes("tableC"),
シリアライズしたputC, putA)


// === ステップ３ ===
// 更新の前提条件チェックと、Aテーブルの更新。
// ここでは、前提条件として、Aテーブルの指定したカラムの更新前の値と
// 現在の値を比較し、それらが同一なら、他のプロセスと競合していないと
// みなして更新する。

boolean tableA_updated = tableA.checkAndPut(更新前の値、putA)

boolean tableB_updated = false;
boolean tableC_updated = false;


// === ステップ４ ===
// 登録しておいた BテーブルとCテーブルの更新内容をすぐに実行する。
if (tableA_updated) {
tableB_updated = wal.processMessage(walB)
tableC_updated = wal.processMessage(walC)
}


// === ステップ５ ===
// もし、BまたはCテーブルの更新に失敗したら、手動でロールバック。
...
...


return (tableA_updated && tableB_updated && tableC_updated)

}

public static class TableBCUpdater implements RowLogMessageListener {

public boolean processMessage(RowLogMessage message) throws ... {

String data = Bytes.toString(message.getData())
Put put = null;
try {
put = message.getPayload() をデシリアライズ
} catch (RowLogException e) {
return false;
}

if ("tableB".equals(data)) {
tableB.put(put);
} else if ("tableC".equals(data)) {
tableC.put(put);
}

return true;
}

}
================================================

updateABC()メソッドのステップ４では、この updateABC()メソッドを実行しているスレッドが、直接、TableBCUpdaterクラスの processMessage()メソッドを実行します。

もし、ステップ３まで実行して、ステップ４の手前でプログラムがこけた場合、Aテーブルだけ更新されていて、BテーブルとCテーブルが更新されていない状態になります（一貫性のない状態）　しかし、バックグラウンドスレッドとして動いている RowLogProcessorが、ステップ２で登録しておいた RowLogMessage を見つけて、TableBCUpdaterに処理させますので、最後にはBとCテーブルも更新され、一貫性が得られます。

もし、ステップ３のAテーブルの checkAndPut() が失敗した場合、A、B、C全てのテーブルが更新されていない状態になります（これはこれで一貫性がある状態）　ステップ２で登録してしまった RowLogMessage がどうなるかというとRowLogProcessorが、putAが未実行なことを検知して実行しないようにしますので大丈夫です。A、B、C間の一貫性は保たれます。

もし、ステップ４を実行しているときに、Bテーブル、または、Cテーブルの更新に失敗した場合、RowLogProcessorに事前に設定しておいた回数だけリトライさせることができます。もし、リトライを諦める場合は、Aテーブルと、BまたはCテーブルの更新内容を、手動にて取り消す必要があります。


BASEトランザクションのメリットは、リソースをロックする範囲が最小で済むため、分散システムに向いていることです。ACIDの場合は、A、B、Cテーブルの該当する行を１度にロックして、２フェーズコミットを実行する必要がありますので、他のプロセスからの get / put をブロックする時間が長くなります。また、デッドロックの心配もあります。一方、BASEでは、明示的なロックは取得せず、A、B、Cデーブルに順番にputしていくので、他のプロセスをブロックする時間は最小で済みます。また、デッドロックの心配もありません。

一方、デメリットは、更新の途中の状態が外部から見えることと、トランザクションが失敗したとき時のロールバックが自動的にはできないことです。なぜ自動的にできないかというと、Bテーブル、Cテーブルの更新が時には非同期で、Aテーブルを更新した時よりずっと後に行われる可能性があるからです。その間に、別のトランザクションが Aテーブルを更新しているかもしれませんので、Aテーブルの内容を単純に昔の状態に戻すわけにはいきません（銀行口座の例ですと、給料が振り込まれてるかも。単純に戻すと、給料がなかったことになってしまいます）　そのため、アプリケーションのビジネスロジックを考慮して、一貫性のある値に設定する必要があります。

> ＞テーブル複数（以下の例ではA,B,Cの3つ）にデータをputする際に
> ＞例えば、A、Bにputした後にCでのputに失敗
> ＞　→A、Bに入れたデータをなかったことにしたい。

>>> （put失敗後に手動でのA,Bのdeleteを行うのは無し）

> 先の例では、上記のように書きましたが、実際は失敗に限らないです。
> A操作後（table.put()等した後）にシステム上のとある条件で登録したデータを
> 戻したいといった場合にどのように対処するのがベストなのか確認したかったです。
> 何かよい案はありますでしょうか？

delete以外ということなら、バージョン管理システム的な考え方で、timestamp を使って１つ前のバージョンを get し、それを最新版として put しなおす。くらいしか思いつかないですね...。もちろん手動ですが。

戻すことは考えないで、逆に、事前にシステム上の条件を全て確認できないでしょうか？　条件がOKなら A を更新。いったん A を更新したら、B と C も必ず更新する、という考え方です。

やっぱり戻したい、ということでしたら、非同期の要素が入る BASEトランザクションだといろいろと面倒だと思います。いま HBase-trx のACIDトランザクションを使っていて性能的に問題ないのでしたら、そのまま HBase-trx を使い続けるのもいいのかもしれません。HBase-trx の HBase 0.89対応版ブランチはもうなくなってますが、git コマンドでリビジョンを指定すれば、0.89に対応していた頃のソースコードを取り出せると思います。