Tablestore：Pemindaian Paralel

Pemindaian Data Menggunakan Satu Utas

Saat menggunakan pemindaian paralel, kode untuk permintaan yang menggunakan satu utas lebih sederhana dibandingkan dengan kode untuk permintaan yang menggunakan beberapa utas. Parameter currentParallelId dan maxParallel tidak diperlukan untuk permintaan yang menggunakan satu utas. Permintaan ParallelScan yang menggunakan satu utas memberikan throughput yang lebih tinggi dibandingkan dengan permintaan Search, namun lebih rendah dibandingkan dengan permintaan ParallelScan yang menggunakan beberapa utas.

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 = "<TableName>";
        String indexName = "<IndexName>";
        // Menanyakan ID sesi dan jumlah maksimum tugas pemindaian paralel yang didukung oleh permintaan.
        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();
        /*
         * Membuat permintaan pemindaian paralel.
         */
        ParallelScanRequest parallelScanRequest = new ParallelScanRequest();
        parallelScanRequest.setTableName(tableName);
        parallelScanRequest.setIndexName(indexName);
        ScanQuery scanQuery = new ScanQuery();
        // Kueri ini menentukan rentang data yang akan dipindai. Anda dapat membuat kueri bersarang dan kompleks.
        Query query = new MatchAllQuery();
        scanQuery.setQuery(query);

        // Tentukan jumlah maksimum baris yang dapat dikembalikan oleh setiap panggilan 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);

        /*
         * Gunakan pembuat untuk membuat permintaan pemindaian paralel yang memiliki fitur yang sama dengan permintaan sebelumnya.
         */
        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<>();

        /*
         * Gunakan API asli untuk memindai data.
         */
        {
            ParallelScanResponse parallelScanResponse = client.parallelScan(parallelScanRequest);
            // Menanyakan token ScanQuery untuk permintaan berikutnya.
            byte[] nextToken = parallelScanResponse.getNextToken();
            // Memperoleh data.
            List<Row> rows = parallelScanResponse.getRows();
            result.addAll(rows);
            while (nextToken != null) {
                // Tentukan token.
                parallelScanRequest.getScanQuery().setToken(nextToken);
                // Lanjutkan memindai data.
                parallelScanResponse = client.parallelScan(parallelScanRequest);
                // Memperoleh data.
                rows = parallelScanResponse.getRows();
                result.addAll(rows);
                nextToken = parallelScanResponse.getNextToken();
            }
        }

        /*
         * Metode yang direkomendasikan.
         * Gunakan iterator untuk memindai semua data yang cocok. Metode ini memiliki kecepatan kueri yang sama tetapi lebih mudah digunakan dibandingkan metode sebelumnya.
         */
        {
            RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
            while (iterator.hasNext()) {
                Row row = iterator.next();
                result.add(row);
                // Memperoleh nilai spesifik.
                String col_1 = row.getLatestColumn("col_1").getValue().asString();
                long col_2 = row.getLatestColumn("col_2").getValue().asLong();
            }
        }

        /*
         * Jika operasi gagal, ulangi operasi. Jika pemanggil fungsi ini memiliki mekanisme ulang atau jika Anda tidak ingin mengulang operasi yang gagal, Anda dapat mengabaikan bagian ini.
         * Untuk memastikan ketersediaan, kami sarankan Anda memulai tugas pemindaian paralel baru saat pengecualian terjadi.
         * Pengecualian berikut mungkin terjadi saat Anda mengirim permintaan ParallelScan:
         * 1. Penyimpangan sesi terjadi di sisi server. Kode kesalahan adalah OTSSessionExpired.
         * 2. Penyimpangan seperti pengecualian jaringan terjadi di sisi klien.
         */
        try {
            // Jalankan logika pemrosesan.
            {
                RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    // Proses baris data. Jika Anda memiliki sumber daya memori yang cukup, Anda dapat menambahkan baris ke daftar.
                    result.add(row);
                }
            }
        } catch (Exception ex) {
            // Ulangi logika pemrosesan.
            {
                result.clear();
                RowIterator iterator = client.createParallelScanIterator(parallelScanRequestByBuilder);
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    // Proses baris data. Jika Anda memiliki sumber daya memori yang cukup, Anda dapat menambahkan baris ke daftar.
                    result.add(row);
                }
            }
        }
        return result;
    }
}

Pemindaian Data Menggunakan Beberapa Utas

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 {
        // Menanyakan jumlah core CPU di klien.
        final int cpuProcessors = Runtime.getRuntime().availableProcessors();
        // Tentukan jumlah utas paralel untuk klien. Kami sarankan Anda menentukan jumlah core CPU di klien sebagai jumlah utas paralel untuk klien untuk mencegah dampak pada kinerja kueri.
        final Semaphore semaphore = new Semaphore(cpuProcessors);

        // Menanyakan ID sesi dan jumlah maksimum tugas pemindaian paralel yang didukung oleh permintaan.
        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();

        // Buat objek AtomicLong jika Anda perlu memperoleh jumlah baris untuk bisnis Anda.
        AtomicLong rowCount = new AtomicLong(0);
        /*
         * Jika Anda ingin melakukan multithreading menggunakan fungsi, Anda dapat membangun kelas internal untuk mewarisi utas.
         * Anda juga dapat membangun kelas eksternal untuk mengatur kode.
         */
        final class ThreadForScanQuery extends Thread {
            private final int currentParallelId;

            private ThreadForScanQuery(int currentParallelId) {
                this.currentParallelId = currentParallelId;
                this.setName("ThreadForScanQuery:" + maxParallel + "-" + currentParallelId);  // Tentukan nama utas.
            }

            @Override
            public void run() {
                System.out.println("memulai utas:" + this.getName());
                try {
                    // Jalankan logika pemrosesan.
                    {
                        ParallelScanRequest parallelScanRequest = ParallelScanRequest.newBuilder()
                                .tableName(tableName)
                                .indexName(indexName)
                                .scanQuery(ScanQuery.newBuilder()
                                        .query(QueryBuilders.range("col_long").lessThan(10_0000)) // Tentukan data yang akan ditanyakan.
                                        .limit(2000)
                                        .currentParallelId(currentParallelId)
                                        .maxParallel(maxParallel)
                                        .build())
                                .addColumnsToGet("col_long", "col_keyword", "col_bool")  // Tentukan bidang yang akan dikembalikan dari indeks pencarian. Untuk mengembalikan semua bidang dari indeks pencarian, atur returnAllColumnsFromIndex ke true.
                                //.returnAllColumnsFromIndex(true)
                                .sessionId(sessionId)
                                .build();
                        // Gunakan iterator untuk memperoleh semua data.
                        RowIterator ltr = client.createParallelScanIterator(parallelScanRequest);
                        long count = 0;
                        while (ltr.hasNext()) {
                            Row row = ltr.next();
                            // Tambahkan logika pemrosesan kustom. Contoh kode berikut menunjukkan cara menambahkan logika pemrosesan kustom untuk menghitung jumlah baris:
                            count++;
                        }
                        rowCount.addAndGet(count);
                        System.out.println("utas[" + this.getName() + "] selesai. utas ini mendapatkan baris:" + count);
                    }
                } catch (Exception ex) {
                    // Jika pengecualian terjadi, Anda dapat mengulang logika pemrosesan.
                } finally {
                    semaphore.release();
                }
            }
        }

        // Eksekusi utas secara simultan. Nilai valid currentParallelId: [0, Nilai maxParallel).
        List<ThreadForScanQuery> threadList = new ArrayList<ThreadForScanQuery>();
        for (int currentParallelId = 0; currentParallelId < maxParallel; currentParallelId++) {
            ThreadForScanQuery thread = new ThreadForScanQuery(currentParallelId);
            threadList.add(thread);
        }

        // Mulai utas secara simultan.
        for (ThreadForScanQuery thread : threadList) {
            // Tentukan nilai untuk semaphore untuk membatasi jumlah utas yang dapat dimulai pada saat yang sama untuk mencegah hambatan di sisi klien.
            semaphore.acquire();
            thread.start();
        }

        // Utas utama diblokir sampai semua utas selesai.
        for (ThreadForScanQuery thread : threadList) {
            thread.join();
        }
        System.out.println("semua utas selesai! total baris:" + rowCount.get());
    }
}