TableStoreWriter のパラメータとコールバック - - Alibaba Cloud ドキュメントセンター

TableStoreWriter を初期化するときに、ビジネス要件に基づいて TableStoreWriter パラメータとコールバック関数ロジックを指定できます。このトピックでは、指定できる TableStoreWriter パラメータについて説明し、コールバックの例を示します。

パラメータ

TableStoreWriter を初期化するときに、WriterConfig を変更して、ビジネス要件に基づいて TableStoreWriter パラメータを指定できます。

例

次のサンプルコードは、WriterConfig を使用して指定できるパラメータの例を示しています。

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);

パラメータ

パラメータ	タイプ	説明
bucketCount	Integer	TableStoreWriter のバケット数。デフォルト値：3。バケットは、データをキャッシュするために使用されるバッファーと同等です。このパラメータを指定して、並列順次書き込みリクエストの数を増やすことができます。マシンのボトルネックに達していない場合、バケットの数は書き込みレートと正の相関があります。バケットの書き込みモードが同時書き込みの場合は、デフォルト設定を保持します。
bufferSize	Integer	メモリ内のバッファキューのサイズ。単位：行。デフォルト値：1024。このパラメータの値は 2 の指数乗の倍数である必要があります。
enableSchemaCheck	Boolean	データをバッファにフラッシュするときにスキーマをチェックするかどうかを指定します。有効な値： true：データをバッファにフラッシュするときにスキーマをチェックします。これはデフォルト値です。行データがバッファにフラッシュされる前に、TableStoreWriter は行データに対して次のチェックを実行します。行のプライマリキースキーマがテーブルに定義されているスキーマと同じかどうかを確認します。行の各プライマリキー列または属性列の値のサイズが制限を超えていないかどうかを確認します。行の属性列の数が制限を超えていないかどうかを確認します。属性列の名前が行のプライマリキー列の名前と同じかどうかを確認します。行のサイズが、リクエストを使用して一度にインポートできる最大データ量を超えていないかどうかを確認します。行データが上記のチェックに失敗した場合、TableStoreWriter は行データがダーティデータであると判断します。ダーティデータはバッファにフラッシュされません。 false：データをバッファにフラッシュするときにスキーマをチェックしません。バッファ内の特定の行データがダーティデータの場合、TableStoreWriter がバッファ内の行データを Tablestore に書き込むときに、行データは Tablestore に書き込まれません。
dispatchMode	DispatchMode	データをバッファにフラッシュするときに、データをバケットにディスパッチするモード。このパラメータは、バケットの数が 2 以上の場合にのみ有効になります。有効な値： HASH_PARTITION_KEY：パーティションキーのハッシュ値に基づいてデータをバケットにディスパッチします。パーティションキーが同じハッシュ値を共有するデータは、同じバケットに順番に書き込まれます。これはデフォルト値です。 HASH_PRIMARY_KEY：プライマリキーのハッシュ値に基づいてデータをバケットにディスパッチします。プライマリキーが同じハッシュ値を共有するデータは、同じバケットに順番に書き込まれます。 ROUND_ROBIN：各バケットをトラバースして、ループでデータをディスパッチします。データは異なるバケットにランダムに分散されます。
batchRequestType	BatchRequestType	TableStoreWriter がバッファ内のデータを Tablestore に書き込むために使用するリクエストのタイプ。有効な値： BATCH_WRITE_ROW：BatchWriteRowRequest。これはデフォルト値です。 BULK_IMPORT：BulkImportRequest。
concurrency	Integer	TableStoreWriter がバッファ内のデータを Tablestore に書き込むために使用する並列リクエストの最大数。デフォルト値：10。
writeMode	WriteMode	TableStoreWriter がバッファ内のデータを Tablestore に書き込むときに、バケット内のデータを Tablestore に書き込むモード。有効な値： PARALLEL：バケットから Tablestore にデータを並列に書き込み、各バケットから Tablestore に同時に書き込みます。これはデフォルト値です。 SEQUENTIAL：バケットから Tablestore にデータを並列に書き込み、各バケットから Tablestore に順番に書き込みます。
allowDuplicatedRowInBatchRequest	Boolean	TableStoreWriter がバッチ書き込みリクエストを作成するときに、同じプライマリキー値を持つ行を許可するかどうかを指定します。デフォルト値：true。データテーブルにセカンダリインデックスが作成されている場合、Tablestore はこのパラメータの設定を無視し、同じプライマリキー値を持つ行を許可しません。この場合、TableStoreWriter はリクエストを作成するときに、同じプライマリキー値を持つ行を異なるリクエストに追加します。
maxBatchSize	Integer	バッチ書き込みリクエストで Tablestore に書き込むことができる最大データ量。単位：バイト。デフォルトでは、バッチ書き込みリクエストで最大 4 MB のデータを Tablestore に書き込むことができます。
maxBatchRowsCount	Integer	バッチ書き込みリクエストで Tablestore に書き込むことができる最大行数。デフォルト値：200。最大値：200。
callbackThreadCount	Integer	TableStoreWriter 内でコールバックを実行するスレッドプールのスレッド数。デフォルト値はプロセッサの数です。
callbackThreadPoolQueueSize	Integer	TableStoreWriter 内でコールバックを実行するスレッドプールのキューサイズ。デフォルト値：1024。
maxColumnsCount	Integer	データをバッファにフラッシュするときの行の最大列数。デフォルト値：128。
maxAttrColumnSize	Integer	データをバッファにフラッシュするときの単一の属性列の値の最大サイズ。デフォルトでは、各属性列の値に最大 2 MB のデータが許可されます。単位：バイト。
maxPKColumnSize	Integer	データをバッファにフラッシュするときの単一のプライマリキー列の値の最大サイズ。デフォルトでは、各プライマリキー列の値に最大 1 KB のデータが許可されます。単位：バイト。
flushInterval	Integer	TableStoreWriter がバッファ内のデータを Tablestore に自動的に書き込む間隔。デフォルト値：10000。単位：ミリ秒。
logInterval	Integer	TableStoreWriter がバッファ内のデータを Tablestore に書き込むときに、タスクステータスが自動的に表示される間隔。デフォルト値：10000。単位：ミリ秒。
clientMaxConnections	Integer	クライアントが内部的に構築されるときに使用される接続の最大数。デフォルト値：300。
writerRetryStrategy	WriterRetryStrategy	クライアントが内部的に構築されるときに使用される再試行ポリシー。有効な値： CERTAIN_ERROR_CODE_NOT_RETRY：特定のエラーコードに対して再試行を試行せず、他のエラーコードに対して再試行を試行します。これはデフォルト値です。再試行が実行されないエラーコード：`OTSParameterInvalid`、`OTSConditionCheckFail`、`OTSRequestBodyTooLarge`、`OTSInvalidPK`、`OTSOutOfColumnCountLimit`、および`OTSOutOfRowSizeLimit` CERTAIN_ERROR_CODE_RETRY：特定のエラーコードに対して再試行を試行し、他のエラーコードに対して再試行を試行しません。再試行が実行されるエラーコード：`OTSInternalServerError`、`OTSRequestTimeout`、`OTSPartitionUnavailable`、`OTSTableNotReady`、`OTSRowOperationConflict`、`OTSTimeout`、`OTSServerUnavailable`、および`OTSServerBusy`

コールバック

TableStoreWriter はコールバックを使用して、書き込みの成功または失敗を報告します。データの行が Tablestore に書き込まれると、TableStoreWriter は onCompleted() 関数を呼び出します。データの行が Tablestore に書き込めなかった場合、TableStoreWriter は例外のカテゴリに基づいて onFailed() 関数を呼び出します。

次のサンプルコードは、コールバックを使用して、Tablestore に書き込まれた行数と Tablestore に書き込めなかった行数の統計を収集する方法の例を示しています。

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();
    }
};