Tablestore:TableStoreWriter を使用してデータを同時書き込みする

背景情報

ロギングや IoT トレースなどのシナリオでは、システムは短期間に大量のデータを生成し、データベースに書き込みます。 データベースは高い書き込み並列性と、毎秒数万行、あるいは数百万行もの高い書き込みスループットを提供する必要があります。 ただし、Tablestore の BatchWriteRow 操作を使用して、バッチで最大 200 行までしか書き込むことができません。

TableStoreWriter は、パフォーマンス専有型で使いやすいデータインポートツールクラスであり、Tablestore SDK for Java によって提供されます。 このクラスは、高並列および高スループットでデータをインポートするために使用される操作をカプセル化します。 TableStoreWriter を使用すると、高並列で Tablestore データテーブルにデータを書き込み、行レベルのコールバックとカスタム構成を指定できます。 詳細については、「付録 1：TableStoreWriter の動作原理」をご参照ください。

シナリオ

以下のビジネス要件がある場合は、TableStoreWriter を使用して、ログストレージ、インスタントメッセージ (IM)、分散キューなどのシナリオで Tablestore にデータを書き込むことができます。

• アプリケーションの高い同時実行性をサポートするために高いスループットが必要です。

• 単一行のデータの書き込みレイテンシに対する要件はありません。

• データは非同期で書き込むことができます（プロデューサー - コンシューマーモードを使用できます）。

• 同じ行のデータを繰り返し書き込むことができます。

前提条件

• Tablestore が有効化され、インスタンスが作成されていること。 詳細については、「Tablestore を有効化してインスタンスを作成する」をご参照ください。

• Alibaba Cloud アカウントまたは Resource Access Management (RAM) ユーザーの AccessKey ペアを取得していること。 詳細については、「AccessKey ペアを作成する」をご参照ください。

手順

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>tablestore</artifactId>
    <version>5.17.4</version>
</dependency>                 

private static TableStoreWriter createTablesStoreWriter() {
    
    /**
     * ほとんどの場合、TableStoreWriter パラメーターのデフォルト値を保持できます。 ビジネス要件に基づいて他の値に設定することもできます。
     * パラメーターの詳細については、「付録 2：TableStoreWriter パラメーター」をご参照ください。
     * */
    WriterConfig config = new WriterConfig();
    // バッチ書き込みリクエストで Tablestore に書き込むことができる行の最大数を指定します。 デフォルト値：200。
    config.setMaxBatchRowsCount(200); 
    // TableStoreWriter がバッファ内のデータを Tablestore に書き込むために使用する並列リクエストの最大数を指定します。 デフォルト値：10。デフォルト値を保持することをお勧めします。                          
    config.setConcurrency(10);    
            
    /**
     * 行レベルのコールバックを構成します。
     * この例では、コールバックは、Tablestore に書き込まれた行数と、Tablestore への書き込みに失敗した行数をカウントするように構成されています。
     * */
    TableStoreCallback<RowChange, RowWriteResult> resultCallback = new TableStoreCallback<RowChange, RowWriteResult>() {
        @Override
        public void onCompleted(RowChange rowChange, RowWriteResult cc) {
            succeedRows.incrementAndGet();
        }

        @Override
        public void onFailed(RowChange rowChange, Exception ex) {
            failedRows.incrementAndGet();
        }
    };

    /** アクセス認証情報を構成します。   **/
    ServiceCredentials credentials = new DefaultCredentials(accessKeyId, accessKeySecret);

    /**
     * 使いやすく、初期化と解放ロジックを分離できる、組み込みのスレッドプールとクライアントを使用することをお勧めします。
     * */
    DefaultTableStoreWriter writer = new DefaultTableStoreWriter(
        endpoint, credentials, instanceName, tableName, config, resultCallback);

    return writer;
}

public void writeSingleRowWithFuture(TableStoreWriter writer) {
    System.out.println("=========================================================[Start]");
    System.out.println("Write Single Row With Future");
    int rowsCount = 1000;
    int columnsCount = 10;
    String strValue = "1234567890";
    AtomicLong rowIndex = new AtomicLong(-1);

    List<Future<WriterResult>> futures = new LinkedList<Future<WriterResult>>();
    for (long index = rowIndex.incrementAndGet(); index < rowsCount; index = rowIndex.incrementAndGet()) {

        PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("pk_0", PrimaryKeyValue.fromString(md5Hex(index + "")))
                .addPrimaryKeyColumn("pk_1", PrimaryKeyValue.fromString("pk" + index))
                .addPrimaryKeyColumn("pk_2", PrimaryKeyValue.fromLong(index % 5))
                .build();

        RowUpdateChange rowChange = new RowUpdateChange(tableName, pk);
        for (int j = 0; j < columnsCount; j++) {
            rowChange.put("column_" + j, ColumnValue.fromString(strValue));
        }
        rowChange.put("index", ColumnValue.fromLong(index));
        Future<WriterResult> future = writer.addRowChangeWithFuture(rowChange);
        futures.add(future);
    }

    System.out.println("Write thread finished.");
    // バッファ内のデータを Tablestore にフラッシュします。 TableStoreWriter は、flushInterval パラメーターと maxBatchSize パラメーターに基づいて、バッファ内のデータを Tablestore にフラッシュするタイミングを決定します。 flushInterval パラメーターは、バッファ内のデータを Tablestore にフラッシュする間隔を指定します。 maxBatchSize パラメーターは、バッファ内のデータ量に基づいてバッファ内のデータを Tablestore にフラッシュするかどうかを指定します。
    writer.flush();
    
    // Future オブジェクトを表示します。
    // printFutureResult(futures);

    System.out.println("=========================================================[Finish]");
}

public void writeRowListWithFuture(TableStoreWriter writer) {
    System.out.println("=========================================================[Start]");
    System.out.println("Write Row List With Future");

    int rowsCount = 1000;
    int columnsCount = 10;
    String strValue = "1234567890";
    AtomicLong rowIndex = new AtomicLong(-1);

    List<Future<WriterResult>> futures = new LinkedList<Future<WriterResult>>();
    List<RowChange> rowChanges = new LinkedList<RowChange>();
    for (long index = rowIndex.incrementAndGet(); index < rowsCount; index = rowIndex.incrementAndGet()) {

        PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("pk_0", PrimaryKeyValue.fromString(md5Hex(index + "")))
                .addPrimaryKeyColumn("pk_1", PrimaryKeyValue.fromString("pk" + index))
                .addPrimaryKeyColumn("pk_2", PrimaryKeyValue.fromLong(index % 5))
                .build();

        RowUpdateChange rowChange = new RowUpdateChange(tableName, pk);
        for (int j = 0; j < columnsCount; j++) {
            rowChange.put("column_" + j, ColumnValue.fromString(strValue));
        }
        rowChange.put("index", ColumnValue.fromLong(index));
        rowChanges.add(rowChange);
        if (Math.random() > 0.995 || index == rowsCount - 1) {
            Future<WriterResult> future = writer.addRowChangeWithFuture(rowChanges);
            futures.add(future);
            rowChanges.clear();
        }
    }

    System.out.println("Write thread finished.");
    // バッファ内のデータを Tablestore にフラッシュします。 TableStoreWriter は、flushInterval パラメーターと maxBatchSize パラメーターに基づいて、バッファ内のデータを Tablestore にフラッシュするタイミングを決定します。 flushInterval パラメーターは、バッファ内のデータを Tablestore にフラッシュする間隔を指定します。 maxBatchSize パラメーターは、バッファ内のデータ量に基づいてバッファ内のデータを Tablestore にフラッシュするかどうかを指定します。
    writer.flush();
    
    // Future オブジェクトを表示します。
    // printFutureResult(futures);
    
    System.out.println("=========================================================[Finish]");
}

// TableStoreWriter を事前にシャットダウンします。 キュー内のすべてのデータが Tablestore に書き込まれた後、システムはクライアントと内部スレッドプールを自動的にシャットダウンします。
writer.close();

完全なサンプルコード

次のサンプルコードは、データテーブルを作成し、テーブルにデータを同時書き込みする方法の例を示しています。

import com.alicloud.openservices.tablestore.DefaultTableStoreWriter;
import com.alicloud.openservices.tablestore.SyncClient;
import com.alicloud.openservices.tablestore.TableStoreCallback;
import com.alicloud.openservices.tablestore.TableStoreWriter;
import com.alicloud.openservices.tablestore.core.auth.DefaultCredentials;
import com.alicloud.openservices.tablestore.core.auth.ServiceCredentials;
import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.writer.RowWriteResult;
import com.alicloud.openservices.tablestore.writer.WriterConfig;
import com.alicloud.openservices.tablestore.writer.WriterResult;

import com.aliyuncs.exceptions.ClientException;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

import static org.apache.commons.codec.digest.DigestUtils.md5Hex;

public class TableStoreWriterDemo {

    // インスタンスの名前を指定します。
    private static String instanceName = "yourInstanceName";
    // インスタンスのエンドポイントを指定します。
    private static String endpoint = "yourEndpoint";
    // 環境変数から AccessKey ID と AccessKey シークレットを取得します。
    private static String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
    private static String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
    private static String tableName = "<TABLE_NAME>";

    private static AtomicLong succeedRows = new AtomicLong();
    private static AtomicLong failedRows = new AtomicLong();

    public static void main(String[] args) throws ClientException {
        TableStoreWriterDemo sample = new TableStoreWriterDemo();

        /**
         * TableStoreWriter を使用する前に、テーブルがすでに存在することを確認してください。
         * 1. TableStoreWriter は、テーブルが存在するかどうかを確認します。
         * 2. 書き込みたいデータのフィールドとフィールドタイプが、テーブルのフィールドとフィールドタイプと同じであるかどうかを確認します。
         * */
        sample.tryCreateTable();

        /**
         * TableStoreWriter を初期化するには、DefaultTablestoreWriter を使用することをお勧めします。
         * DefaultTableStoreWriter(
         *      String endpoint,                                                   // インスタンスのエンドポイント。
         *      ServiceCredentials credentials,                                    // アクセス認証情報（AccessKey ペアを含む）。 トークンがサポートされています。
         *      String instanceName,                                               // インスタンスの名前。
         *      String tableName,                                                  // テーブルの名前。 TableStoreWriter は 1 つのテーブルにのみデータを書き込むことができます。
         *      WriterConfig config,                                               // TableStoreWriter の構成。
         *      TableStoreCallback<RowChange, RowWriteResult> resultCallback       // 行レベルのコールバック。
         * )
         * */
        TableStoreWriter writer = sample.createTablesStoreWriter();

        /**
         * Future オブジェクトでは、一度に 1 行のデータを書き込むメソッドが使用されます。
         * */
        sample.writeSingleRowWithFuture(writer);
        /**
         * Future オブジェクトでは、一度に複数行のデータを書き込むメソッドが使用されます。
         * */   
        //sample.writeRowListWithFuture(writer);

        System.out.println("Count by TablestoreCallback: failedRow=" + failedRows.get() + ", succeedRow=" + succeedRows.get());
        System.out.println("Count by WriterStatics: " + writer.getWriterStatistics());

        /**
         * TableStoreWriter を事前にシャットダウンします。 キュー内のすべてのデータが Tablestore に書き込まれた後、システムはクライアントと内部スレッドプールを自動的にシャットダウンします。
         * */
        writer.close();
    }

    private static TableStoreWriter createTablesStoreWriter() {

        WriterConfig config = new WriterConfig();
        // バッチ書き込みリクエストで Tablestore に書き込むことができる行の最大数を指定します。 デフォルト値：200。バッチ書き込みリクエストで 200 行を超えるデータを書き込みたい場合は、このパラメーターにより大きな値を指定します。
        config.setMaxBatchRowsCount(200); 
        // TableStoreWriter がバッファ内のデータを Tablestore に書き込むために使用する並列リクエストの最大数を指定します。 デフォルト値：10。デフォルト値を保持することをお勧めします。                          
        config.setConcurrency(10);                                   

        /**
         * 行レベルのコールバックを構成します。
         * この例では、コールバックは、Tablestore に書き込まれた行数と、Tablestore への書き込みに失敗した行数をカウントするように構成されています。
         * */
        TableStoreCallback<RowChange, RowWriteResult> resultCallback = new TableStoreCallback<RowChange, RowWriteResult>() {
            @Override
            public void onCompleted(RowChange rowChange, RowWriteResult cc) {
                succeedRows.incrementAndGet();
            }

            @Override
            public void onFailed(RowChange rowChange, Exception ex) {
                failedRows.incrementAndGet();
            }
        };

        ServiceCredentials credentials = new DefaultCredentials(accessKeyId, accessKeySecret);


        /**
         * 使いやすく、初期化と解放ロジックを分離できる、組み込みのスレッドプールとクライアントを使用することをお勧めします。
         * */
        DefaultTableStoreWriter writer = new DefaultTableStoreWriter(
                endpoint, credentials, instanceName, tableName, config, resultCallback);

        return writer;
    }


    private static void tryCreateTable() throws ClientException {
        SyncClient ots = new SyncClient(endpoint, accessKeyId, accessKeySecret, instanceName);

        try {
            ots.deleteTable(new DeleteTableRequest(tableName));
        } catch (Exception e) {
        }

        TableMeta tableMeta = new TableMeta(tableName);
        tableMeta.addPrimaryKeyColumn("pk_0", PrimaryKeyType.STRING);
        tableMeta.addPrimaryKeyColumn("pk_1", PrimaryKeyType.STRING);
        tableMeta.addPrimaryKeyColumn("pk_2", PrimaryKeyType.INTEGER);
        TableOptions tableOptions = new TableOptions(-1, 1);
        CreateTableRequest request = new CreateTableRequest(
                tableMeta, tableOptions, new ReservedThroughput(new CapacityUnit(0, 0)));

        try {
            CreateTableResponse res = ots.createTable(request);
        } catch (Exception e) {
            throw new ClientException(e);
        } finally {
            ots.shutdown();
        }
    }

    public static void writeSingleRowWithFuture(TableStoreWriter writer) {
        System.out.println("=========================================================[Start]");
        System.out.println("Write Single Row With Future");
        int rowsCount = 1000;
        int columnsCount = 10;
        String strValue = "1234567890";
        AtomicLong rowIndex = new AtomicLong(-1);

        List<Future<WriterResult>> futures = new LinkedList<Future<WriterResult>>();
        for (long index = rowIndex.incrementAndGet(); index < rowsCount; index = rowIndex.incrementAndGet()) {

            PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                .addPrimaryKeyColumn("pk_0", PrimaryKeyValue.fromString(md5Hex(index + "")))
                .addPrimaryKeyColumn("pk_1", PrimaryKeyValue.fromString("pk" + index))
                .addPrimaryKeyColumn("pk_2", PrimaryKeyValue.fromLong(index % 5))
                .build();

            RowUpdateChange rowChange = new RowUpdateChange(tableName, pk);
            for (int j = 0; j < columnsCount; j++) {
                rowChange.put("column_" + j, ColumnValue.fromString(strValue));
            }
            rowChange.put("index", ColumnValue.fromLong(index));
            Future<WriterResult> future = writer.addRowChangeWithFuture(rowChange);
            futures.add(future);
        }

        System.out.println("Write thread finished.");
        writer.flush();
        // Future オブジェクトを表示します。
        // printFutureResult(futures);

        System.out.println("=========================================================[Finish]");
    }
    
    public void writeRowListWithFuture(TableStoreWriter writer) {
        System.out.println("=========================================================[Start]");
        System.out.println("Write Row List With Future");

        int rowsCount = 1000;
        int columnsCount = 10;
        String strValue = "1234567890";
        AtomicLong rowIndex = new AtomicLong(-1);

        List<Future<WriterResult>> futures = new LinkedList<Future<WriterResult>>();
        List<RowChange> rowChanges = new LinkedList<RowChange>();
        for (long index = rowIndex.incrementAndGet(); index < rowsCount; index = rowIndex.incrementAndGet()) {

            PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
                    .addPrimaryKeyColumn("pk_0", PrimaryKeyValue.fromString(md5Hex(index + "")))
                    .addPrimaryKeyColumn("pk_1", PrimaryKeyValue.fromString("pk" + index))
                    .addPrimaryKeyColumn("pk_2", PrimaryKeyValue.fromLong(index % 5))
                    .build();

            RowUpdateChange rowChange = new RowUpdateChange(tableName, pk);
            for (int j = 0; j < columnsCount; j++) {
                rowChange.put("column_" + j, ColumnValue.fromString(strValue));
            }
            rowChange.put("index", ColumnValue.fromLong(index));
            rowChanges.add(rowChange);
            if (Math.random() > 0.995 || index == rowsCount - 1) {
                Future<WriterResult> future = writer.addRowChangeWithFuture(rowChanges);
                futures.add(future);
                rowChanges.clear();
            }
    }

    System.out.println("Write thread finished.");
    writer.flush();
    // Future オブジェクトを表示します。
    // printFutureResult(futures);
    System.out.println("=========================================================[Finish]");
    }


    private static void printFutureResult(List<Future<WriterResult>> futures) {
        int totalRow = 0;

        for (int index = 0; index < futures.size(); index++) {
            try {
                WriterResult result = futures.get(index).get();
                totalRow += result.getTotalCount();
                System.out.println(String.format(
                        "Future[%d] finished:\tfailed: %d\tsucceed: %d\tfutureBatch: %d\ttotalFinished: %d",
                        index, result.getFailedRows().size(), result.getSucceedRows().size(),
                        result.getTotalCount(), totalRow));

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }
}

この例では、次の結果が返されます。

=========================================================[Start]
Write Single Row With Future
Write thread finished.
=========================================================[Finish]
Count by TablestoreCallback: failedRow=0, succeedRow=1000
Count by WriterStatics: WriterStatistics: {
    totalRequestCount=6,
    totalRowsCount=1000,
    totalSucceedRowsCount=1000,
    totalFailedRowsCount=0,
    totalSingleRowRequestCount=0,
}

FAQ

Tablestore SDK for Java を使用して Tablestore データテーブルにデータを書き込むときに、「属性列の数が最大値 128 を超えています」というエラーが報告された場合はどうすればよいですか？

付録

WriterConfig config = new WriterConfig();
config.setBucketCount(3);
config.setBufferSize(1024);
config.setEnableSchemaCheck(true);
config.setDispatchMode(DispatchMode.HASH_PARTITION_KEY);
config.setBatchRequestType(BatchRequestType.BATCH_WRITE_ROW);
config.setConcurrency(10);
config.setWriteMode(WriteMode.PARALLEL);
config.setAllowDuplicatedRowInBatchRequest(true);
config.setMaxBatchSize(4 * 1024 * 1024);
config.setMaxBatchRowsCount(200);
config.setCallbackThreadCount(16);
config.setCallbackThreadPoolQueueSize(1024);
config.setMaxColumnsCount(128);
config.setMaxAttrColumnSize(2 * 1024 * 1024);
config.setMaxPKColumnSize(1024);
config.setFlushInterval(10000);
config.setLogInterval(10000);
config.setClientMaxConnections(300);
config.setWriterRetryStrategy(WriterRetryStrategy.CERTAIN_ERROR_CODE_NOT_RETRY);

private static AtomicLong succeedRows = new AtomicLong();
private static AtomicLong failedRows = new AtomicLong();
TableStoreCallback<RowChange, RowWriteResult> resultCallback = new TableStoreCallback<RowChange, RowWriteResult>() {
    @Override
    public void onCompleted(RowChange rowChange, RowWriteResult cc) {
        // Tablestore に書き込まれた行数をカウントします。
        succeedRows.incrementAndGet();
    }

    @Override
    public void onFailed(RowChange rowChange, Exception ex) {
        // Tablestore への書き込みに失敗した行数をカウントします。
        failedRows.incrementAndGet();
    }
};