インメモリデータグリッド（Apache Ignite）をDB（PostgreSQL）へ永続化する

はじめに

前回はApache Igniteをキャッシュサーバ（インメモリKVS）として利用してみました。
今回はデータをRDB（PostgreSQL）へ永続化してみます。

Apache Ingiteのサイトの図では以下に相当します。

画像の出典：Apache Ignite Database Caching

Write Throught（ライトスルー）、Read Through（リードスルー）がポイントです。

リードスルー、ライトスルーについて

リードスルーにより、データがキャッシュから要求され、存在しない場合は自動的にデータベースからデータを読み込みます。
ライトスルーでは、インメモリキャッシュで更新が発生すると永続化対象のデータベースに反映できます。

ただし、ライトスルーでは各キャッシュのputおよびremove操作にて、永続化対象のDBへの更新が発生するため、キャッシュ更新の実行時間が比較的長くなり、ストレージへの負荷も高まる可能性があります。このようなケースを軽減するために、Igniteでは更新を累積し非同期に永続対象のDBの更新を実行します。（また、ライトバックとも呼ばれます）。

RDBへの永続化の種類

RDBへの永続化の方法としては以下の3つがあります。
今回は簡単な設定で永続化できるCacheJdbcBlobStoreを利用します。

• CacheJdbcBlobStore CacheJdbcBlobStoreは、データベースに自動でENTRIESテーブルを作成し、データを保存します。テーブルには主キーのakeyカラムとvalカラムがあり、BLOB形式でvalカラムに保存されます。
• CacheJdbcPojoStore CacheJdbcPojoStoreは、データ（キャッシュ）をPOJOで扱い、テーブルへマッピングし保存します。例えば、POJOで3つのフィールド（A, B, C）があれば、テーブルも3つのカラム（A, B, C）があることになります。
• CacheHibernateBlobStore CacheHibernateBlobStoreは、Hibernateによってサポートされているそうです（が興味がないので調べていません）

環境

環境は前回から引き続き、VagrantでCentOS 7（1台構成）をインストールした環境を利用します。
Apache Igniteは「/opt/apache-ignite」以下にインストール済みであることを前提とします。

IgniteとPostgreSQLは同一サーバ（192.168.10.71）に起動しています。

PostgreSQLをインストールする

以下を参考にPostgreSQL 9.6をインストールしました。

• CentOS7にPostgreSQLを構築する方法
• CentOSにPostgreSQLをインストールする

PostgreSQLのJDBCドライバが必要ですので、以下からダウンロードし、「/opt/apache-ignite-fabric-2.6.0-bin/libs/」に格納します。

Apache IgniteにPostgreSQLへの接続と永続化の設定を行う

Igniteの設定ファイルとして、以下のファイルを作成します。

/opt/apache-ignite/config/default-config-postgres.xml

/opt/apache-ignite/config/default-config-postgres.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean id="grid.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
        <property name="discoverySpi">
            <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
                <property name="ipFinder">
                    <bean
                        class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">
                        <property name="addresses">
                            <list>
                                <value>192.168.10.71:47500..47509</value>
                            </list>
                        </property>
                    </bean>
                </property>
            </bean>
        </property>

        <property name="cacheConfiguration">
            <list>
                <bean class="org.apache.ignite.configuration.CacheConfiguration">
                    <constructor-arg name="name" value="testCache"></constructor-arg>
                    <property name="cacheStoreFactory">
                        <bean
                            class="org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStoreFactory">
                            <property name="dataSourceBean" value="postgresDataSouorceTestdb" />
                        </bean>
                    </property>
                    <property name="writeThrough" value="true" />
                    <property name="readThrough" value="true" />
                </bean>
            </list>
        </property>
    </bean>

    <bean id="postgresDataSouorceTestdb" class="org.postgresql.ds.PGPoolingDataSource">
        <property name="url" value="jdbc:postgresql://192.168.10.71:5432/testdb" />
        <property name="user" value="postgres" />
        <property name="password" value="postgres" />
    </bean>
</beans>

「<bean class="org.apache.ignite.configuration.CacheConfiguration">」でキャッシュの設定を追加しています。この設定がない場合、クライアントからキャッシュが要求されると自動でキャッシュが作成されます。今回は、永続化の設定を実施するため、あらかじめ設定ファイルにキャッシュの設定を記載しています。

• property name="cacheStoreFactory"：CacheJdbcBlobStoreFactoryを指定し、データソースに「postgresDataSouorceTestdb」を指定
• property name="writeThrough"：　ライトスルーを有効化
• property name="readThrough"：　リードスルーを有効化
• bean id="postgresDataSouorceTestdb"：　PostgreSQLのデータソースを定義。DBは"testdb"、user/passwordは"postgres"にしています。

次にIgniteへアクセスするJavaプログラムを作成します。コードは「CACHE_NAME = "testCache"」が異なるだけで前回とほぼ同様です。CACHE_NAMEには「default-config-postgres.xml」に設定したキャッシュの名前とあわせる必要があります。

import org.apache.ignite.Ignition;
import org.apache.ignite.client.ClientCache;
import org.apache.ignite.client.ClientException;
import org.apache.ignite.client.IgniteClient;
import org.apache.ignite.configuration.ClientConfiguration;

public class PostgresCachePersistApp {

    public static void main(String[] args) {
        // 接続先のApache IgniteサーバのIPとポートを指定する。
        ClientConfiguration cfg = new ClientConfiguration().setAddresses("192.168.10.71:10800");

        try (IgniteClient igniteClient = Ignition.startClient(cfg)) {
            final String CACHE_NAME = "testCache";

            ClientCache<Integer, String> cache = igniteClient.getOrCreateCache(CACHE_NAME);

            Integer key = 1;
            String val = "put-get-test";

            cache.put(key, val);

            System.out.format("PUT [%s]\n", val);

            String cachedVal = cache.get(key);

            System.out.format("GET [%s]\n", cachedVal);
        }
        catch (ClientException e) {
            e.printStackTrace();
        }
        catch (Exception e) {
            e.printStackTrace();
        }
    }
}

実行すると、以下のエラーが発生します。
Igniteで自動で永続化のためのテーブルを作成しているのですが、PostgreSQLではテーブル作成に失敗するようです。

[02:28:14,034][SEVERE][client-connector-#116][ClientListenerNioListener] Failed to process client request [req=o.a.i.i.processors.platform.client.cache.ClientCachePutRequest@e60ac9e]
javax.cache.integration.CacheWriterException: class org.apache.ignite.internal.processors.cache.CachePartialUpdateCheckedException: Failed to update keys (retry update if possible).: [1]
～～省略～～
                Caused by: class org.apache.ignite.IgniteException: Failed to create database table.
                        at org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStore.init(CacheJdbcBlobStore.java:432)
                        at org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStore.write(CacheJdbcBlobStore.java:247)
                        at org.apache.ignite.internal.processors.cache.store.GridCacheStoreManagerAdapter.put(GridCacheStoreManagerAdapter.java:586)
                        ... 36 more
                Caused by: org.postgresql.util.PSQLException: ERROR: type "binary" does not exist
  Position: 42
                        at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2440)
                        at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2183)
                        at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:308)
                        at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:441)
                        at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:365)
                        at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:307)
                        at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:293)
                        at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:270)
                        at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:266)
                        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
                        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
                        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
                        at java.lang.reflect.Method.invoke(Method.java:498)
                        at org.postgresql.ds.PGPooledConnection$StatementHandler.invoke(PGPooledConnection.java:428)
                        at com.sun.proxy.$Proxy29.execute(Unknown Source)
                        at org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStore.init(CacheJdbcBlobStore.java:425)
                        ... 38 more

テーブル作成用のSQLはプロパティで設定できるので、「<bean
class="org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStoreFactory">」以下を追記し再度実行します。

<property name="createTableQuery" value="create table if not exists ENTRIES (akey bytea primary key, val bytea)" />

「select count(*) from entries」でレコード数をカウントすると、"1"となっており永続化できていることを確認できます。（A5:SQL Mk-2を使用しています）
また、ここには記載しませんでしたがIgniteを再起動しキャッシュがクリアされても、GETするとDBに永続化されたデータを取得できることも確認しました（当たり前ですが）。

リードスルー、ライトスルー、ライトビハインドの設定

リードスルー、ライトスルー、ライトビハインドの設定について、もう少し補足します。

#	設定項目	説明	デフォルト値
1	readThrough	リードスルーが有効・無効を設定する。	false
2	writeThrough	ライトスルーが有効・無効を設定します。writeBehindEnabledを有効にする場合は、この値も有効にします。	false
3	writeBehindEnabled	ライトビハンドが有効・無効を設定します。	false
4	writeBehindFlushSize	ライトビハンドキャッシュの最大サイズを設定します。ライトビハンドキャッシュのサイズがこの値を超過すると、キャッシュされたすべてのデータがCacheStoreへフラッシュされ、キャッシュがクリアされます。この値が0の場合は、writeBehindFlushFrequencyに従ってフラッシュが実行されます。writeBehindFlushSizeとwriteBehindFlushFrequencyの両方を0に設定することはできません。	10240(byte)
5	writeBehindFlushFrequency	ライトビハンドキャッシュがCacheStoreにフラッシュされる間隔（ミリ秒）を設定します。この値は、キャッシュへのデータのput/removeがCacheStoreへ適用されるまでの最大間隔を定義します。この値が0の場合、writeBehindFlushSizeに従ってフラッシュが実行されます。	5000(ms)
6	writeBehindFlushThreadCount	ライトビハンドキャッシュのフラッシュを実行するスレッドの数を設定します。	1
7	writeBehindBatchSize	ライトビハンドのCacheStore操作の最大バッチサイズを設定する。	512

https://apacheignite.readme.io/docs/3rd-party-store#section-cachejdbcblobstore 参照

設定のサンプルを以下に記載します。

                                <bean class="org.apache.ignite.configuration.CacheConfiguration">
                                        <constructor-arg name="name" value="testCache"></constructor-arg>
                                        <property name="cacheStoreFactory">
                                                <bean
                                                        class="org.apache.ignite.cache.store.jdbc.CacheJdbcBlobStoreFactory">
                                                        <property name="dataSourceBean" value="postgresDataSouorceTestdb" />
                                                        <property name="createTableQuery"
                                                                value="create table if not exists ENTRIES (akey bytea primary key, val bytea)" />
                                                </bean>
                                        </property>
                                        <property name="writeThrough" value="true" />
                                        <property name="readThrough" value="true" />
                                        <property name="writeBehindEnabled" value="true" />
                                        <property name="writeBehindFlushSize" value="10240" />
                                        <property name="writeBehindFlushFrequency" value="5000" />
                                        <property name="writeBehindFlushThreadCount" value="1" />
                                        <property name="writeBehindBatchSize" value="512" />
                                </bean>

おわりに

今回、インメモリKVS（キャッシュサーバ）をDBに永続化しましたが、DBへはバイナリで保存されているため、DBをSELECTしてもどのようなデータが保存されているのかが分かりません。簡単な設定で永続化できるのは良いのですが、バイナリで保存されるためサイズも大きなっている（はず）です。