Tablestore:データの読み取り

パラメータ

例

データバージョンと読み取る列を指定し、フィルターまたは正規表現を使用してデータをフィルタリングできます。

private static void getRow(SyncClient client, String pkValue) {
    // 主キーを構築します。
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString(pkValue));
    PrimaryKey primaryKey = primaryKeyBuilder.build();

    // テーブル名と主キーを指定して、データの行を読み取ります。
    SingleRowQueryCriteria criteria = new SingleRowQueryCriteria("<TABLE_NAME>", primaryKey);
    // MaxVersions パラメータを 1 に設定して、データの最新バージョンを読み取ります。
    criteria.setMaxVersions(1);
    GetRowResponse getRowResponse = client.getRow(new GetRowRequest(criteria));
    Row row = getRowResponse.getRow();

    System.out.println("読み取り完了。結果：");
    System.out.println(row);

    // 読み取る列を指定します。
    criteria.addColumnsToGet("Col0");
    getRowResponse = client.getRow(new GetRowRequest(criteria));
    row = getRowResponse.getRow();

    System.out.println("読み取り完了。結果：");
    System.out.println(row);
} 

private static void getRow(SyncClient client, String pkValue) {
    // 主キーを構築します。
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString(pkValue));
    PrimaryKey primaryKey = primaryKeyBuilder.build();

    // テーブル名と主キーを指定して、データの行を読み取ります。
    SingleRowQueryCriteria criteria = new SingleRowQueryCriteria("<TABLE_NAME>", primaryKey);
    // MaxVersions パラメータを 1 に設定して、データの最新バージョンを読み取ります。
    criteria.setMaxVersions(1);

    // フィルターを構成して、Col0 列の値が 0 である行を返します。
    SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("Col0",
            SingleColumnValueFilter.CompareOperator.EQUAL, ColumnValue.fromLong(0));
    // Col0 列が存在しない場合、行は返されません。
    singleColumnValueFilter.setPassIfMissing(false);
    criteria.setFilter(singleColumnValueFilter);

    GetRowResponse getRowResponse = client.getRow(new GetRowRequest(criteria));
    Row row = getRowResponse.getRow();

    System.out.println("読み取り完了。結果：");
    System.out.println(row);
}

private static void getRow(SyncClient client, String pkValue) {
    // データテーブルの名前を指定します。
    SingleRowQueryCriteria criteria = new SingleRowQueryCriteria("<TABLE_NAME>");

    // 主キーを構築します。
    PrimaryKey primaryKey = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString(pkValue))
        .build();
    criteria.setPrimaryKey(primaryKey);

    // MaxVersions パラメータを 1 に設定して、データの最新バージョンを読み取ります。
    criteria.setMaxVersions(1);

    // フィルターを構成します。 cast<int>(regex(Col1)) が 100 より大きい場合、行が返されます。
    RegexRule regexRule = new RegexRule("t1:([0-9]+),", RegexRule.CastType.VT_INTEGER);
    SingleColumnValueRegexFilter filter =  new SingleColumnValueRegexFilter("Col1",
        regexRule,SingleColumnValueRegexFilter.CompareOperator.GREATER_THAN, ColumnValue.fromLong(100));
    criteria.setFilter(filter);

    GetRowResponse getRowResponse = client.getRow(new GetRowRequest(criteria));
    Row row = getRowResponse.getRow();

    System.out.println("読み取り完了。結果：");
    System.out.println(row);
}

使用上の注意

• BatchGetRow 操作では、すべての行に同じパラメーター構成が使用されます。たとえば、ColumnsToGet パラメーターが [colA] に設定されている場合、すべての行から colA 列の値のみが読み取られます。

• BatchGetRow 操作を呼び出して複数の行を同時に読み取ると、一部の行の読み取りに失敗する場合があります。この場合、Tablestore は例外を返さず、失敗した行のエラーメッセージを含む BatchGetRowResponse を返します。そのため、BatchGetRow 操作を呼び出す場合は、戻り値を確認する必要があります。BatchGetRowResponse の isAllSucceed メソッドを使用して、すべての行が読み取られたかどうかを確認するか、BatchGetRowResponse の getFailedRows メソッドを使用して、失敗した行に関する情報を取得できます。

• BatchGetRow 操作を呼び出して、最大 100 行を同時に読み取ることができます。

パラメーター

詳細については、このトピックの1 つのデータ行を読み取るためのパラメーターセクションをご参照ください。

例

次のサンプルコードは、10 行のデータを読み取るためのバージョン条件、読み取る列、およびフィルターを構成する方法の例を示しています。

private static void batchGetRow(SyncClient client) {
    // データテーブルの名前を指定します。
    MultiRowQueryCriteria multiRowQueryCriteria = new MultiRowQueryCriteria("<TABLE_NAME>");
    // 読み取る 10 行を指定します。
    for (int i = 0; i < 10; i++) {
        PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
        primaryKeyBuilder.addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("pk" + i));
        PrimaryKey primaryKey = primaryKeyBuilder.build();
        multiRowQueryCriteria.addRow(primaryKey);
    }
    // 条件を追加します。
    multiRowQueryCriteria.setMaxVersions(1);
    multiRowQueryCriteria.addColumnsToGet("Col0");
    multiRowQueryCriteria.addColumnsToGet("Col1");
    SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("Col0",
            SingleColumnValueFilter.CompareOperator.EQUAL, ColumnValue.fromLong(0));
    singleColumnValueFilter.setPassIfMissing(false);
    multiRowQueryCriteria.setFilter(singleColumnValueFilter);

    BatchGetRowRequest batchGetRowRequest = new BatchGetRowRequest();
    // BatchGetRow 操作では、複数のテーブルからデータを読み取ることができます。各 multiRowQueryCriteria パラメーターは、1 つのテーブルのクエリ条件を指定します。複数の multiRowQueryCriteria パラメーターを追加して、複数のテーブルからデータを読み取ることができます。
    batchGetRowRequest.addMultiRowQueryCriteria(multiRowQueryCriteria);

    BatchGetRowResponse batchGetRowResponse = client.batchGetRow(batchGetRowRequest);

    System.out.println("すべての操作が成功したかどうか: " + batchGetRowResponse.isAllSucceed());
    System.out.println("読み取り完了。結果: ");
    for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getSucceedRows()) {
        System.out.println(rowResult.getRow());
    }
    if (!batchGetRowResponse.isAllSucceed()) {
        for (BatchGetRowResponse.RowResult rowResult : batchGetRowResponse.getFailedRows()) {
            System.out.println("失敗した行: " + batchGetRowRequest.getPrimaryKey(rowResult.getTableName(), rowResult.getIndex()));
            System.out.println("失敗の原因: " + rowResult.getError());
        }

        /**
         * createRequestForRetry メソッドを使用して、失敗した行の操作を再試行するための別のリクエストを構成できます。この例では、再試行リクエストのみが構成されています。
         * 再試行方法として、Tablestore SDK のカスタム再試行ポリシーを使用することをお勧めします。このようにして、バッチ操作の実行後に失敗した行を再試行できます。再試行ポリシーを指定した後、操作を呼び出すために再試行コードを追加する必要はありません。
         */
        BatchGetRowRequest retryRequest = batchGetRowRequest.createRequestForRetry(batchGetRowResponse.getFailedRows());
    }
}

詳細なサンプルコードについては、BatchGetRow@ GitHub をご覧ください。

使用上の注意

GetRange 操作は、左端一致の原則に従います。 Tablestore は、プライマリキー値が指定された範囲内にあるデータを読み取るために、最初のプライマリキー列から最後のプライマリキー列までの値を順番に比較します。 たとえば、データテーブルのプライマリキーは、PK1、PK2、PK3 というプライマリキー列で構成されています。 データが読み取られると、Tablestore はまず、行の PK1 値が最初のプライマリキー列に指定された範囲内にあるかどうかを判断します。 行の PK1 値が範囲内にある場合、Tablestore は、行の他のプライマリキー列の値が各プライマリキー列に指定された範囲内にあるかどうかの判断を停止し、行を返します。 行の PK1 値が範囲内にない場合、Tablestore は、PK1 と同じ方法で、行の他のプライマリキー列の値が各プライマリキー列に指定された範囲内にあるかどうかを判断し続けます。

次のいずれかの条件が満たされると、GetRange 操作が停止してデータが返される場合があります。

• スキャンされたデータの量が 4 MB に達する。

• スキャンされた行数が 5,000 に達する。

• 返される行数が上限に達する。

• 予約済みの読み取りスループットがすべて消費されているため、読み取りスループットが次のデータ行を読み取るのに不十分である。

各 GetRange 呼び出しはデータを 1 回スキャンします。 GetRange 操作を呼び出してスキャンするデータのサイズが大きい場合、スキャンされた行数が 5,000 に達するか、スキャンされたデータのサイズが 4 MB に達すると、スキャンは停止します。 Tablestore は、クエリ条件を満たす残りのデータを返しません。 ページングメソッドを使用して、クエリ条件を満たす残りのデータを取得できます。

パラメーター

例

private static void getRange(SyncClient client, String startPkValue, String endPkValue) {
    // データテーブルの名前を指定します。
    RangeRowQueryCriteria rangeRowQueryCriteria = new RangeRowQueryCriteria("<TABLE_NAME>");

    // 開始プライマリキーを指定します。
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString(startPkValue));
    rangeRowQueryCriteria.setInclusiveStartPrimaryKey(primaryKeyBuilder.build());

    // 終了プライマリキーを指定します。
    primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString(endPkValue));
    rangeRowQueryCriteria.setExclusiveEndPrimaryKey(primaryKeyBuilder.build());

    rangeRowQueryCriteria.setMaxVersions(1);

    System.out.println("GetRange result:");
    while (true) {
        GetRangeResponse getRangeResponse = client.getRange(new GetRangeRequest(rangeRowQueryCriteria));
        for (Row row : getRangeResponse.getRows()) {
            System.out.println(row);
        }

        // nextStartPrimaryKey パラメーターの値が null でない場合は、読み取り操作を続行します。
        if (getRangeResponse.getNextStartPrimaryKey() != null) {
            rangeRowQueryCriteria.setInclusiveStartPrimaryKey(getRangeResponse.getNextStartPrimaryKey());
        } else {
            break;
        }
    }
}         

private static void getRange(SyncClient client, String startPkValue, String endPkValue) {
    // データテーブルの名前を指定します。
    RangeRowQueryCriteria rangeRowQueryCriteria = new RangeRowQueryCriteria("<TABLE_NAME>");
    // 開始プライマリキーを指定します。 この例では、2 つのプライマリキー列が使用されています。
    PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk1", PrimaryKeyValue.fromString(startPkValue));// 最初のプライマリキー列の値を特定の値に設定します。
    primaryKeyBuilder.addPrimaryKeyColumn("pk2", PrimaryKeyValue.INF_MIN);// 2 番目のプライマリキー列の値を無限に小さい値に設定します。
    rangeRowQueryCriteria.setInclusiveStartPrimaryKey(primaryKeyBuilder.build());

    // 終了プライマリキーを指定します。
    primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
    primaryKeyBuilder.addPrimaryKeyColumn("pk1", PrimaryKeyValue.fromString(endPkValue));// 最初のプライマリキー列の値を特定の値に設定します。
    primaryKeyBuilder.addPrimaryKeyColumn("pk2", PrimaryKeyValue.INF_MAX);// 2 番目のプライマリキー列の値を無限に大きい値に設定します。
    rangeRowQueryCriteria.setExclusiveEndPrimaryKey(primaryKeyBuilder.build());

    rangeRowQueryCriteria.setMaxVersions(1);

    System.out.println("GetRange result:");
    while (true) {
        GetRangeResponse getRangeResponse = client.getRange(new GetRangeRequest(rangeRowQueryCriteria));
        for (Row row : getRangeResponse.getRows()) {
            System.out.println(row);
        }

        // nextStartPrimaryKey パラメーターの値が null でない場合は、読み取り操作を続行します。
        if (getRangeResponse.getNextStartPrimaryKey() != null) {
            rangeRowQueryCriteria.setInclusiveStartPrimaryKey(getRangeResponse.getNextStartPrimaryKey());
        } else {
            break;
        }
    }
}    

private static void getRange(SyncClient client) {
    // データテーブルの名前を指定します。
    RangeRowQueryCriteria criteria = new RangeRowQueryCriteria("<TABLE_NAME>");

    // 読み取るデータのプライマリキーの範囲として ["pk:2020-01-01.log", "pk:2021-01-01.log") を指定します。 範囲は左閉右開区間です。
    PrimaryKey pk0 = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("2020-01-01.log"))
        .build();
    PrimaryKey pk1 = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("2021-01-01.log"))
        .build();
    criteria.setInclusiveStartPrimaryKey(pk0);
    criteria.setExclusiveEndPrimaryKey(pk1);

    // MaxVersions パラメーターを 1 に設定して、データの最新バージョンを読み取ります。
    criteria.setMaxVersions(1);

    // フィルターを設定します。 cast<int>(regex(Col1)) が 100 より大きい場合、行が返されます。
    RegexRule regexRule = new RegexRule("t1:([0-9]+),", RegexRule.CastType.VT_INTEGER);
    SingleColumnValueRegexFilter filter =  new SingleColumnValueRegexFilter("Col1",
        regexRule,SingleColumnValueRegexFilter.CompareOperator.GREATER_THAN,ColumnValue.fromLong(100));
    criteria.setFilter(filter);

    while (true) {
        GetRangeResponse resp = client.getRange(new GetRangeRequest(criteria));
        for (Row row : resp.getRows()) {
            // 何かをする
            System.out.println(row);
        }
        if (resp.getNextStartPrimaryKey() != null) {
            criteria.setInclusiveStartPrimaryKey(resp.getNextStartPrimaryKey());
        } else {
            break;
        }
   }
}

詳細なサンプルコードについては、「GetRange@ GitHub」をご参照ください。

はじめに

1. TableStoreReader を構築します。

   // インスタンス名を指定します。
   String instanceName = "yourInstanceName";
   // インスタンスのエンドポイントを指定します。
   String endpoint = "yourEndpoint";
   // 環境変数から AccessKey ID と AccessKey シークレットを取得します。
   String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
   String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
   
   AsyncClientInterface client = new AsyncClient(endpoint, accessKeyId, accessKeySecret, instanceName);
   TableStoreReaderConfig config = new TableStoreReaderConfig();
   ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(1024))
   
   TableStoreReader reader = new DefaultTableStoreReader(client, config, executor, null);

2. クエリリクエストを構築します。

   クエリするデータをメモリにキャッシュします。一度に 1 行以上のデータを追加できます。

   PrimaryKey pk1 = PrimaryKeyBuilder.createPrimaryKeyBuilder()
           .addPrimaryKeyColumn("pk1", PrimaryKeyValue.fromLong(0))
           .addPrimaryKeyColumn("pk2", PrimaryKeyValue.fromLong(0))
           .build();
   // 値が pk1 のプライマリキーを追加して、行の属性列をクエリします。
   Future<ReaderResult> readerResult = reader.addPrimaryKeyWithFuture("<TABLE_NAME1>", pk1);
   // List メソッドを使用して、複数のプライマリキー値を同時に追加して、行をクエリすることもできます。
   List<PrimaryKey> primaryKeyList = new ArrayList<PrimaryKey>();
   Future<ReaderResult> readerResult = reader.addPrimaryKeysWithFuture("<TABLE_NAME2>", primaryKeyList);

3. データをクエリします。

   メモリにキャッシュされているデータをクエリするリクエストを送信します。同期モードまたは非同期モードでデータをクエリできます。

   • 同期モードでデータをクエリする

     reader.flush();

   • 非同期モードでデータをクエリする

     reader.send();

4. クエリ結果を取得します。

   // 成功したクエリと失敗したクエリに関する情報を表示します。
   for (RowReadResult success : readerResult.get().getSucceedRows()) {
       System.out.println(success.getRowResult());
   }
   
   for (RowReadResult fail : readerResult.get().getFailedRows()) {
       System.out.println(fail.getRowResult());
   }

5. TableStoreReader を閉じます。

   reader.close();
   // ビジネス要件に基づいて、クライアントとエグゼキュータを閉じます。
   client.shutdown();
   executor.shutdown();

パラメーター

TableStoreReaderConfig を変更して、TableStoreReader のカスタム構成を指定できます。

クエリ条件を指定する

データバージョンの最大数、クエリする列、データをクエリする時間範囲など、テーブルレベルのパラメーターを指定してデータをクエリできます。

// 過去 60 秒以内に指定されたテーブルの col1 列の最大 10 バージョンのデータをクエリします。
// データテーブルの名前を指定します。
RowQueryCriteria criteria = new RowQueryCriteria("<TABLE_NAME>");
// 返す列を指定します。
criteria.addColumnsToGet("col1");
// 返すバージョンの最大数を指定します。
criteria.setMaxVersions(10);
criteria.setTimeRange(new TimeRange(System.currentTimeMillis() - 60 * 1000, System.currentTimeMillis()));
reader.setRowQueryCriteria(criteria);

完全なサンプルコード

public class TableStoreReaderDemo {
    // インスタンス名を指定します。
    private static final String instanceName = "yourInstanceName";
    // インスタンスのエンドポイントを指定します。
    private static final String endpoint = "yourEndpoint";
    // 環境変数から AccessKey ID と AccessKey シークレットを取得します。
    private static final String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
    private static final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
    private static AsyncClientInterface client;
    private static ExecutorService executor;
    private static AtomicLong succeedRows = new AtomicLong();
    private static AtomicLong failedRows = new AtomicLong();

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        /**
         * ステップ 1: TableStoreReader を構築します。
         */
        // AsyncClient を構築します。
        client = new AsyncClient(endpoint, accessKeyId, accessKeySecret, instanceName);
        // TableStoreReader の構成クラスを構築します。
        TableStoreReaderConfig config = new TableStoreReaderConfig();
        {
            // 以下のパラメーターにはデフォルト値があり、空のままにすることができます。
            // クエリする行のプライマリキー値を追加する前に、テーブルのスキーマを確認します。
            config.setCheckTableMeta(true);
            // リクエストを使用してクエリできる行の最大数を指定します。 この例では、リクエストを使用して最大 100 行をクエリできます。
            config.setMaxBatchRowsCount(100);
            // 返されるバージョンの最大数を指定します。
            config.setDefaultMaxVersions(1);
            // 送信できる同時リクエストの総数。
            config.setConcurrency(16);
            // メモリ内のバケットの数を指定します。
            config.setBucketCount(4);
            // キャッシュされたすべてのデータがフラッシュされる間隔を指定します。
            config.setFlushInterval(10000);
            // TableStoreReader のステータスが記録される間隔を指定します。
            config.setLogInterval(10000);
        }
        // リクエストの送信に使用するエグゼキュータを構築します。
        ThreadFactory threadFactory = new ThreadFactory() {
            private final AtomicInteger counter = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "reader-" + counter.getAndIncrement());
            }
        };
        executor = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue(1024), threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());

        // TableStoreReader のコールバック関数を構築します。
        TableStoreCallback<PrimaryKeyWithTable, RowReadResult> callback = new TableStoreCallback<PrimaryKeyWithTable, RowReadResult>() {
            @Override
            public void onCompleted(PrimaryKeyWithTable req, RowReadResult res) {
                succeedRows.incrementAndGet();
            }

            @Override
            public void onFailed(PrimaryKeyWithTable req, Exception ex) {
                failedRows.incrementAndGet();
            }
        };
        TableStoreReader reader = new DefaultTableStoreReader(client, config, executor, callback);
        /**
         * ステップ 2: リクエストを構築します。
         */
        // クエリする行のプライマリキー値をメモリに追加します。
        PrimaryKey pk1 = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("pk1", PrimaryKeyValue.fromLong(0))
                .addPrimaryKeyColumn("pk2", PrimaryKeyValue.fromLong(0))
                .build();
        reader.addPrimaryKey("<TABLE_NAME1>", pk1);

        // クエリする別の行のプライマリキー値をメモリに追加し、Future オブジェクトを取得します。
        PrimaryKey pk2 = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("pk1", PrimaryKeyValue.fromLong(0))
                .addPrimaryKeyColumn("pk2", PrimaryKeyValue.fromLong(0))
                .build();
        Future<ReaderResult> readerResult = reader.addPrimaryKeyWithFuture("<TABLE_NAME2>", pk2);
        /**
         * ステップ 3: データをクエリします。
         */
        // メモリ内のデータを非同期モードで送信します。
        reader.send();
        /**
         * ステップ 4: クエリ結果を取得します。
         */
        // 成功したクエリと失敗したクエリに関する情報を表示します。
        for (RowReadResult success : readerResult.get().getSucceedRows()) {
            System.out.println(success.getRowResult());
        }
        for (RowReadResult fail : readerResult.get().getFailedRows()) {
            System.out.println(fail.getRowResult());
        }
        /**
         * ステップ 5: TableStoreReader を閉じます。
         */
        reader.close();
        client.shutdown();
        executor.shutdown();
    }
}