Aerospikeことはじめ ～JavaからAerospikeに接続・CRUD操作～

はじめに

前回はローカルマシンにAerospikeを(Dockerイメージで)入れたので
今回はJavaからAerospikeにアクセスしてみる。

環境

ソフトウェア	バージョン
OS	Windows10
Docker Engine	20.10.5
Java	11

AerospikeにJavaでアクセスするには
Javaのバージョンが8以上である必要がある。

準備

1. Javaクライアントライブラリの取得

まず初めにAerospikeのJavaクライアントライブラリを入手する。
MavenやGradle、Ivyで入手可能。
今回はGradleを使用。
build.gradleに以下の依存ライブラリ定義を追加。

build.gradle

repositories {
    mavenCentral()
}

dependencies {
    compile "com.aerospike:aerospike-client:5.0.0"
}

この定義をいれてGradleのbuildを実行すれば
ライブラリがDLされる。

2. データの準備

AerospikeのDockerコンテナに入り、
AQL(Aerospike Query Language) でデータを入れる。
（AerospikeのDockerコンテナ、AQLについては、前回の記事参照）

データ挿入

aql> INSERT INTO test.NintenGames (PK, Name, Release) VALUES ('SMB', 'SuperMarioBros', 1983)
OK, 1 record affected.

aql> INSERT INTO test.NintenGames (PK, Name, Release) VALUES ('LOZ', 'LegendOfZelda', 1986)
OK, 1 record affected.

aql> INSERT INTO test.NintenGames (PK, Name, Release) VALUES ('PKM', 'Pockemon', 1998)
OK, 1 record affected.

aql> SELECT Name, Release FROM test.NintenGames
+------------------+----------+
| Name             | Release  |
+------------------+----------+
| "LegendOfZelda"  | 1986     |
| "SuperMarioBros" | 1983     |
| "Pockemon"       | 1998     |
+------------------+----------+
3 rows in set (0.101 secs)

OK

こんな感じのデータができた。

実装

1. 接続

接続方法は下記のとおり。

単一のノードを指定する方法と
複数のノードを指定する方法がある。
複数ノード指定した場合は、指定されたホストのどれかに接続する。
一つも接続できなかった場合はAerospikeExceptionがスローされる。

// 単一ノード指定
AerospikeClient client = new AerospikeClient("127.0.0.1", 3000);

// 複数ノード指定
Host[] hosts = new Host[] {
    new Host("127.0.0.1", 3000),
    new Host("another.host", 3000),
    new Host("and.another.host", 3000)
};

ClientPolicy clPolicy = new ClientPolicy();
AerospikeClient clientMultiTarget = new AerospikeClient(clPolicy, hosts);

接続ポートはデフォルトでは3000番
Aerospikeサーバ側の以下の設定ファイルで確認したり設定変更が可能。
/etc/aerospike/aerospike.conf

また、
TLS接続や接続時のユーザ認証などもできるが今回は省略。
（ユーザ認証はEnterpriseEditionのみ）

2. トランザクション

2-1. 挿入

Recordの挿入は以下のような流れ。

1. 書き込みポリシー生成
2. キー生成
3. Binオブジェクト生成
4. 挿入(client.put呼び出し)

挿入コード

// ①書き込みポリシー生成
WritePolicy wPolicy = new WritePolicy();
wPolicy.timeoutDelay = 50;  // 50 millisecond timeout.
wPolicy.expiration = 100;   // 100 seconds TTL.

// ②キーを生成
// 引数をNamespace(DBに相当)、Set(Tableに相当)、PKの順に指定する
Key key = new Key("test", "NintenGames", "KBY");

// ③書き込むBin(Columnに相当)を生成
Bin binName = new Bin("Name", "Kirby");
Bin binRelease = new Bin("Release", 1992);

// ④挿入
client.put(wPolicy, key, binName, binRelease);

①では、書き込みポリシー設定として、接続タイムアウトやRecordのTTL(expiration)などを設定できる。
TTLは秒で指定。0を指定した場合、Namespaceの"default-ttl"の値が設定される。
期限無しにしたい場合は、-1を指定する。
他にも色々設定できる。APIリファレンス参照。

②では、Namespace(DBに相当)、Set(Tableに相当)、PKを持つKeyオブジェクトを生成。
このKeyオブジェクトでどこに挿入するかを指定するわけだ。

③では、レコードの値となるBin(Columnに相当)を生成。

④で挿入される。第3引数以降はBin型の可変長引数になっているのでいくつでも指定可能。

このコード実行後のDB内容は以下の通り。

DB内容

aql> SELECT Name, Release FROM test.NintenGames
+------------------+----------+
| Name             | Release  |
+------------------+----------+
| "LegendOfZelda"  | 1986     |
| "SuperMarioBros" | 1983     |
| "Kirby"          | 1992     |
| "Pockemon"       | 1998     |
+------------------+----------+
4 rows in set (0.103 secs)

Kirbyが追加された！⭐

2-2. 参照

Recordの参照は以下のような流れ。

1. 読み込みポリシー生成
2. キー生成
3. Recordを取得
4. RecordからBinの値を取得

参照コード

// ①読み込みポリシー指定
ScanPolicy sPolicy = new ScanPolicy();
sPolicy.timeoutDelay = 50;  // 50 millisecond timeout.

// ②キーを生成
// 引数をnamespace(DBに相当)、set(Tableに相当)、PKの順に指定する
Key key2 = new Key("test", "NintenGames", "LOZ");

// ③Recordを取得
Record record = client.get(sPolicy, key2);

// ④RecordからBinの値を取得
System.out.printf("Name=[%s] Release=[%d]", record.getString("Name"), record.getInt("Release"));

①では、読み込みポリシー設定として、接続タイムアウトや1秒あたりの読み込みレコード制限数などを指定できる。

②は書き込み時と同様、参照したいRecordのキーを生成。

③でRecordを取得。

④でRecordオブジェクトのget〇〇メソッドでBinの値を取得。
get〇〇の引数にはBin名を指定。
〇〇の部分はINSERT時に入れたデータの型と一致したメソッドを呼ばないと例外が発生する。

このコード実行結果は以下の通り。

実行結果

Name=[LegendOfZelda] Release=[1986]

LegendOfZeldaの情報が取れた！🗡️

2-3. 削除

削除は、削除したいキー情報を指定するだけ。

削除コード

// ①削除対象キー生成
Key key3 = new Key("test", "NintenGames", "PKM");

// ②削除
client.delete(wPolicy, key3);

このコード実行後のDB内容は以下の通り。

aql> SELECT Name, Release FROM test.NintenGames
+------------------+----------+
| Name             | Release  |
+------------------+----------+
| "LegendOfZelda"  | 1986     |
| "SuperMarioBros" | 1983     |
| "Kirby"          | 1992     |
+------------------+----------+
3 rows in set (0.111 secs)

OK

Pockemonが削除された！⭕

2-4. 複数Record参照

複数Recordを取得したい場合は下記のようにする。

複数Record取得コード

// Batchポリシー生成
BatchPolicy bPolicy = new BatchPolicy();

// 参照したいキーを生成
Key[] keys = new Key[3];
keys[0] =  new Key("test", "NintenGames", "SMB");
keys[1] =  new Key("test", "NintenGames", "LOZ");
keys[2] =  new Key("test", "NintenGames", "KBY");

// 複数のRecordを取得
Arrays.stream(client.get(bPolicy, keys))
      .forEach(rec -> System.out.printf("Name=[%s] Release=[%d]\n", rec.getString("Name"), rec.getInt("Release")));

Batchポリシーはサーバーへのリクエストを並列マルチスレッドで行う等を指定できるらしい。（デフォルトは1スレッド）

このコード実行結果は以下の通り。

実行結果

Name=[SuperMarioBros] Release=[1983]
Name=[LegendOfZelda] Release=[1986]
Name=[Kirby] Release=[1992]

2-5. 複合操作

1度のクエリで複数の操作を送ることができる。

複合操作コード

// ①複合操作対象キー生成
Key key4 = new Key("test", "NintenGames", "SMB");

// ②複合更新Bin
Bin binUpdateName = new Bin("Name", "SuperMarioBros2");
Bin binUpdateRelease = new Bin("Release", 3);

// ③複合操作実行
Record record2 = client.operate(
        wPolicy,
        key4,
        Operation.put(binUpdateName),
        Operation.add(binUpdateRelease),
        Operation.get());

// ④RecordからBinの値を取得
System.out.printf("Name=[%s] Release=[%d]\n", record2.getString("Name"), record2.getInt("Release"));

③の複合操作では、client.operateメソッドを呼び出し、

• 値の更新(Operation.put)
• 値のインクリメント(Operation.add)
• Recordの取得(Operation.get)

を一度に実行している。
なお、第一引数のポリシーは書き込みポリシー(WritePolicy)を渡す必要がある。

実行結果は以下のとおり。

実行結果

Name=[SuperMarioBros2] Release=[1986]

実行前DB

aql> SELECT Name, Release FROM test.NintenGames
+------------------+----------+
| Name             | Release  |
+------------------+----------+
| "LegendOfZelda"  | 1986     |
| "SuperMarioBros" | 1983     |
| "Pockemon"       | 1998     |
+------------------+----------+
3 rows in set (0.103 secs)

OK

実行後DB

aql> SELECT Name, Release FROM test.NintenGames
+-------------------+----------+
| Name              | Release  |
+-------------------+----------+
| "LegendOfZelda"   | 1986     |
| "SuperMarioBros2" | 1986     |
| "Pockemon"        | 1998     |
+-------------------+----------+
3 rows in set (0.094 secs)

OK

SuperMarioBrosがSuperMarioBros2になってる！✨

2-6. 切断

最後にクライアントを切断する。

切断コード

client.close();

おわりに

今回はとりあえずここまで。
これでJavaからAerospikeの簡単なCRUD操作はできるようになった。
RecordとBinの関係を理解すれば、
操作は結構シンプルだと思う。

参考

Aerospike Javaクライアント
https://docs.aerospike.com/docs/client/java/index.html

APIリファレンス
https://docs.aerospike.com/apidocs/java/