スループットを最適化した書き込み

このページでは、Spanner における書き込みのスループットを最適化するために、commit（書き込み）の最大遅延時間を設定する方法について説明します。

概要

データの整合性を確保するために、Spanner はデータベース内のすべての投票レプリカに書き込みリクエストを送信します。このレプリケーション プロセスには、計算のオーバーヘッドが発生する可能性があります。詳細については、レプリケーションをご覧ください。

スループットが最適化された書き込みには、一群の書き込みを同時に実行してこうした計算コストを平均化するオプションがあります。このオプションを実行すると、Spanner ではわずかな遅延が発生し、同じ投票参加者に送信する必要がある一群の書き込みがまとめられます。この方法で書き込みを実行すると、レイテンシが若干増加するものの、スループットは大幅に向上します。

デフォルトの動作

commit の遅延時間を設定しない場合、Spanner によって書き込みのコストが平均化すると判断されるとわずかな遅延が設定されることがあります。

一般的なユースケース

書き込みリクエストの遅延時間は、アプリケーションのニーズに応じて手動で設定できます。commit の最大遅延時間を 0 ミリ秒に設定して、レイテンシの影響を受けやすいアプリケーションの commit の遅延を無効にすることもできます。

レイテンシの許容度が高いアプリケーションでスループットを最適化する場合、commit の遅延時間を長く設定すると、各書き込みのレイテンシが大きくなる一方でスループットが大幅に向上します。たとえば、大量のデータを一括で読み込む際に Spanner が個々のデータを書き込む速度が、アプリケーションにとって気にするものでない場合は、commit 遅延時間をより長い値（100 ミリ秒など）に設定できます。最初は 100 ミリ秒の値から始めて、レイテンシとスループットのトレードオフでニーズが満たされるまで増減して調整することをおすすめします。ほとんどのアプリケーションでは、20～100 ミリ秒の値が最適です。

レイテンシの影響を受けやすいアプリケーションの場合、Spanner もデフォルトではレイテンシの影響を受けます。ただし、ワークロードの急増がある場合、Spanner はわずかな遅延を設定することがあります。値を 0 ミリ秒に設定してテストすると、スループットは低下してもレイテンシの低減がアプリケーションにとって妥当なかどうかを判断できます。

commit 遅延時間を複数の値で設定する

書き込みのサブセットに対してさまざまな最大 commit 遅延時間を構成できます。これを行うと、Spanner は一連の書き込みで構成されている最短の遅延時間を使用します。ただし、ほとんどのユースケースにおいて 1 つの値を選択することをおすすめします。その方が挙動の予測を立てやすいためです。

制限事項

commit の遅延時間は 0～500 ミリ秒の間で設定できます。commit の遅延を 500 ミリ秒より長く設定するとエラーが発生します。

commit リクエストに最大 commit 遅延を設定する

最大 commit 遅延のパラメータは CommitRequest メソッドの一部です。このメソッドには、RPC API、REST API、または Cloud Spanner クライアント ライブラリを使用してアクセスできます。

C#


using Google.Cloud.Spanner.Data;
using System;
using System.Threading.Tasks;

public class CommitDelayAsyncSample
{
    public async Task<int> CommitDelayAsync(string projectId, string instanceId, string databaseId)
    {
        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId}";

        using var connection = new SpannerConnection(connectionString);
        await connection.OpenAsync();

        return await connection.RunWithRetriableTransactionAsync(async transaction =>
        {
            transaction.TransactionOptions.MaxCommitDelay = TimeSpan.FromMilliseconds(100);

            using var insertSingerCmd = connection.CreateInsertCommand("Singers",
                new SpannerParameterCollection
                {
                    { "SingerId", SpannerDbType.Int64, 1 },
                    { "FirstName", SpannerDbType.String, "Marc" },
                    { "LastName", SpannerDbType.String, "Richards" }
                });
            insertSingerCmd.Transaction = transaction;
            int rowsInserted = await insertSingerCmd.ExecuteNonQueryAsync();

            using var insertAlbumCmd = connection.CreateInsertCommand("Albums",
                new SpannerParameterCollection
                {
                    { "SingerId", SpannerDbType.Int64, 1 },
                    { "AlbumId", SpannerDbType.Int64, 2 },
                    { "AlbumTitle", SpannerDbType.String, "Go, Go, Go" }
                });
            insertAlbumCmd.Transaction = transaction;
            rowsInserted += await insertAlbumCmd.ExecuteNonQueryAsync();

            return rowsInserted;
        });
    }
}

Go


import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/spanner"
)

func setMaxCommitDelay(w io.Writer, db string) error {
	// db is the fully-qualified database name of the form `projects/<project>/instances/<instance-id>/database/<database-id>`
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return fmt.Errorf("setMaxCommitDelay.NewClient: %w", err)
	}
	defer client.Close()

	commitDelay := 100 * time.Millisecond
	resp, err := client.ReadWriteTransactionWithOptions(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		stmt := spanner.Statement{
			SQL: `INSERT Singers (SingerId, FirstName, LastName)
					VALUES (111, 'Virginia', 'Watson')`,
		}
		rowCount, err := txn.Update(ctx, stmt)
		if err != nil {
			return err
		}
		fmt.Fprintf(w, "%d record(s) inserted.\n", rowCount)
		return nil
	}, spanner.TransactionOptions{CommitOptions: spanner.CommitOptions{MaxCommitDelay: &commitDelay, ReturnCommitStats: true}})
	if err != nil {
		return fmt.Errorf("setMaxCommitDelay.ReadWriteTransactionWithOptions: %w", err)
	}
	fmt.Fprintf(w, "%d mutations in transaction\n", resp.CommitStats.MutationCount)
	return nil
}

Java


import com.google.cloud.spanner.CommitResponse;
import com.google.cloud.spanner.DatabaseClient;
import com.google.cloud.spanner.DatabaseId;
import com.google.cloud.spanner.Mutation;
import com.google.cloud.spanner.Options;
import com.google.cloud.spanner.Spanner;
import com.google.cloud.spanner.SpannerOptions;
import java.time.Duration;
import java.util.Arrays;

public class SetMaxCommitDelaySample {

  static void setMaxCommitDelay() {
    // TODO(developer): Replace these variables before running the sample.
    final String projectId = "my-project";
    final String instanceId = "my-instance";
    final String databaseId = "my-database";

    try (Spanner spanner =
        SpannerOptions.newBuilder().setProjectId(projectId).build().getService()) {
      final DatabaseClient databaseClient = spanner
          .getDatabaseClient(DatabaseId.of(projectId, instanceId, databaseId));
      setMaxCommitDelay(databaseClient);
    }
  }

  static void setMaxCommitDelay(DatabaseClient databaseClient) {
    final CommitResponse commitResponse = databaseClient.writeWithOptions(Arrays.asList(