Tablestore:並列スキャン

単一スレッドを使用したデータのスキャン

並列スキャンを使用する場合、単一スレッドを使用するリクエストのコードは、複数スレッドを使用するリクエストのコードよりも簡単です。単一スレッドを使用するリクエストには、currentParallelId パラメータと maxParallel パラメータは必要ありません。単一スレッドを使用する ParallelScan リクエストは、Search リクエストよりも高いスループットを提供します。ただし、単一スレッドを使用する ParallelScan リクエストは、複数スレッドを使用する ParallelScan リクエストよりも低いスループットを提供します。

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.model.ComputeSplitsRequest;
import com.alicloud.openservices.tablestore.model.ComputeSplitsResponse;
import com.alicloud.openservices.tablestore.model.Row;
import com.alicloud.openservices.tablestore.model.SearchIndexSplitsOptions;
import com.alicloud.openservices.tablestore.model.iterator.RowIterator;
import com.alicloud.openservices.tablestore.model.search.ParallelScanRequest;
import com.alicloud.openservices.tablestore.model.search.ParallelScanResponse;
import com.alicloud.openservices.tablestore.model.search.ScanQuery;
import com.alicloud.openservices.tablestore.model.search.SearchRequest.ColumnsToGet;
import com.alicloud.openservices.tablestore.model.search.query.MatchAllQuery;
import com.alicloud.openservices.tablestore.model.search.query.Query;
import com.alicloud.openservices.tablestore.model.search.query.QueryBuilders;

public class Test {

    public static List<Row> scanQuery(final SyncClient client) {
        String tableName = "<テーブル名>"; // テーブル名を指定します。
        String indexName = "<インデックス名>"; // インデックス名を指定します。
        // セッション ID とリクエストでサポートされる並列スキャンタスクの最大数をクエリします。
        ComputeSplitsRequest computeSplitsRequest = new ComputeSplitsRequest();
        computeSplitsRequest.setTableName(tableName);
        computeSplitsRequest.setSplitsOptions(new SearchIndexSplitsOptions(indexName));
        ComputeSplitsResponse computeSplitsResponse = client.computeSplits(computeSplitsRequest);
        byte[] sessionId = computeSplitsResponse.getSessionId();
        int splitsSize = computeSplitsResponse.getSplitsSize();
        /*
         * 並列スキャンリクエストを作成します。
         */
        ParallelScanRequest parallelScanRequest = new ParallelScanRequest();
        parallelScanRequest.setTableName(tableName);
        parallelScanRequest.setIndexName(indexName);
        ScanQuery scanQuery = new ScanQuery();
        // このクエリは、スキャンするデータの範囲を決定します。ネストされた複雑なクエリを作成できます。
        Query query = new MatchAllQuery();
        scanQuery.setQuery(query);

        // 各 ParallelScan 呼び出しで返される行の最大数を指定します。
        scanQuery.setLimit(2000);
        parallelScanRequest.setScanQuery(scanQuery);
        ColumnsToGet columnsToGet = new ColumnsToGet();
        columnsToGet.setColumns(Arrays.asList("col_1", "col_2"));
        parallelScanRequest.setColumnsToGet(columnsToGet);
        parallelScanRequest.setSessionId(sessionId);

        /*
         * builder を使用して、前のリクエストと同じ機能を持つ並列スキャンリクエストを作成します。
         */
        ParallelScanRequest parallelScanRequestByBuilder = ParallelScanRequest.newBuilder()
            .tableName(tableName)
            .indexName(indexName)
            .scanQuery(ScanQuery.newBuilder()
                .query(QueryBuilders.matchAll())
                .limit(2000)
                .build())
            .addColumnsToGet("col_1", "col_2")
            .sessionId(sessionId)
            .build();
        List<Row> result = new ArrayList<>();

        /*
         * ネイティブ API オペレーションを使用してデータをスキャンします。
         */
        {
            ParallelScanResponse parallelScanResponse = client.parallelScan(parallelScanRequest);
            // 次のリクエストの ScanQuery のトークンをクエリします。
            byte[] nextToken = parallelScanResponse.getNextToken();
            // データを取得します。
            List<Row> rows = parallelScanResponse.getRows();
            result.addAll(rows);
            while (nextToken != null) {
                // トークンを指定します。
                parallelScanRequest.getScanQuery().setToken(nextToken);
                // データのスキャンを続けます。
                parallelScanResponse = client.parallelScan(parallelScanRequest);
                // データを取得します。
                rows = parallelScanResponse.getRows();
                result.addAll(rows);
                nextToken = parallelScanResponse.getNextToken();
            }
        }

        /*
         * 推奨される方法。
         * イテレータを使用して、一致するすべてのデータをスキャンします。この方法は、前の方法と比較して、同じクエリ速度ですが、使いやすくなっています。
         */
        {
            RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
            while (iterator.hasNext()) {
                Row row = iterator.next();
                result.add(row);
                // 特定の値を取得します。
                String col_1 = row.getLatestColumn("col_1").getValue().asString();
                long col_2 = row.getLatestColumn("col_2").getValue().asLong();
            }
        }

        /*
         * オペレーションが失敗した場合は、オペレーションを再試行します。この関数の呼び出し元に再試行メカニズムがある場合、または失敗したオペレーションを再試行したくない場合は、この部分を無視できます。
         * 可可用性を確保するために、例外が発生した場合は新しい並列スキャンタスクを開始することをお勧めします。
         * ParallelScan リクエストを送信すると、次の例外が発生する可能性があります。
         * 1. サーバー側でセッション例外が発生します。エラーコードは OTSSessionExpired です。
         * 2. クライアント側でネットワーク例外などの例外が発生します。
         */
        try {
            // 処理ロジックを実行します。
            {
                RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    // データ行を処理します。十分なメモリリソースがある場合は、行をリストに追加できます。
                    result.add(row);
                }
            }
        } catch (Exception ex) {
            // 処理ロジックを再試行します。
            {
                result.clear();
                RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    // データ行を処理します。十分なメモリリソースがある場合は、行をリストに追加できます。
                    result.add(row);
                }
            }
        }
        return result;
    }
}

複数スレッドを使用したデータのスキャン

import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.model.ComputeSplitsRequest;
import com.alicloud.openservices.tablestore.model.ComputeSplitsResponse;
import com.alicloud.openservices.tablestore.model.Row;
import com.alicloud.openservices.tablestore.model.SearchIndexSplitsOptions;
import com.alicloud.openservices.tablestore.model.iterator.RowIterator;
import com.alicloud.openservices.tablestore.model.search.ParallelScanRequest;
import com.alicloud.openservices.tablestore.model.search.ScanQuery;
import com.alicloud.openservices.tablestore.model.search.query.QueryBuilders;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;

public class Test {

    public static void scanQueryWithMultiThread(final SyncClient client, String tableName, String indexName) throws InterruptedException {
        // クライアントの CPU コア数をクエリします。
        final int cpuProcessors = Runtime.getRuntime().availableProcessors();
        // クライアントの並列スレッド数を指定します。クエリのパフォーマンスへの影響を防ぐために、クライアントの CPU コア数をクライアントの並列スレッド数として指定することをお勧めします。
        final Semaphore semaphore = new Semaphore(cpuProcessors);

        // セッション ID とリクエストでサポートされる並列スキャンタスクの最大数をクエリします。
        ComputeSplitsRequest computeSplitsRequest = new ComputeSplitsRequest();
        computeSplitsRequest.setTableName(tableName);
        computeSplitsRequest.setSplitsOptions(new SearchIndexSplitsOptions(indexName));
        ComputeSplitsResponse computeSplitsResponse = client.computeSplits(computeSplitsRequest);
        final byte[] sessionId = computeSplitsResponse.getSessionId();
        final int maxParallel = computeSplitsResponse.getSplitsSize();

        // ビジネスで行数を取得する必要がある場合は、AtomicLong オブジェクトを作成します。
        AtomicLong rowCount = new AtomicLong(0);
        /*
         * 関数を使用してマルチスレッドを実行する場合は、スレッドを継承する内部クラスを構築できます。
         * また、コードを整理するために外部クラスを構築することもできます。
         */
        final class ThreadForScanQuery extends Thread {
            private final int currentParallelId;

            private ThreadForScanQuery(int currentParallelId) {
                this.currentParallelId = currentParallelId;
                this.setName("ThreadForScanQuery:" + maxParallel + "-" + currentParallelId);  // スレッド名を指定します。
            }

            @Override
            public void run() {
                System.out.println("スレッドを開始します:" + this.getName());
                try {
                    // 処理ロジックを実行します。
                    {
                        ParallelScanRequest parallelScanRequest = ParallelScanRequest.newBuilder()
                                .tableName(tableName)
                                .indexName(indexName)
                                .scanQuery(ScanQuery.newBuilder()
                                        .query(QueryBuilders.range("col_long").lessThan(10_0000)) // クエリするデータを指定します。
                                        .limit(2000)
                                        .currentParallelId(currentParallelId)
                                        .maxParallel(maxParallel)
                                        .build())
                                .addColumnsToGet("col_long", "col_keyword", "col_bool")  // 検索インデックスから返すフィールドを指定します。検索インデックスからすべてのフィールドを返すには、returnAllColumnsFromIndex を true に設定します。
                                //.returnAllColumnsFromIndex(true)
                                .sessionId(sessionId)
                                .build();
                        // イテレータを使用してすべてのデータを取得します。
                        RowIterator ltr = client.createParallelScanIterator(parallelScanRequest);
                        long count = 0;
                        while (ltr.hasNext()) {
                            Row row = ltr.next();
                            // カスタム処理ロジックを追加します。次のサンプルコードは、行数をカウントするためのカスタム処理ロジックを追加する方法を示しています。
                            count++;
                        }
                        rowCount.addAndGet(count);
                        System.out.println("スレッド[" + this.getName() + "] が完了しました。このスレッドが取得した行:" + count);
                    }
                } catch (Exception ex) {
                    // 例外が発生した場合は、処理ロジックを再試行できます。
                } finally {
                    semaphore.release();
                }
            }
        }

        // currentParallelId の有効な値: [0, maxParallel の値)。同時にスレッドを実行します。
        List<ThreadForScanQuery> threadList = new ArrayList<ThreadForScanQuery>();
        for (int currentParallelId = 0; currentParallelId < maxParallel; currentParallelId++) {
            ThreadForScanQuery thread = new ThreadForScanQuery(currentParallelId);
            threadList.add(thread);
        }

        // スレッドを同時に開始します。
        for (ThreadForScanQuery thread : threadList) {
            // セマフォの値を指定して、同時に開始できるスレッドの数を制限し、クライアントのボトルネックを防ぎます。
            semaphore.acquire();
            thread.start();
        }

        // すべてのスレッドが完了するまで、メインスレッドはブロックされます。
        for (ThreadForScanQuery thread : threadList) {
            thread.join();
        }
        System.out.println("すべてのスレッドが完了しました! 合計行数:" + rowCount.get());
    }
}