jdbcを使用してSpannerに行をバッチ挿入する際の読み込みパフォーマンスが低い
背景：TSV形式のデータファイル（MySQLデータベースからダンプされたもの）をGCPSpannerテーブルに読み込もうとしています。
- クライアントライブラリ：公式のSpannerJDBC依存関係v1.15.0
- テーブルスキーマ：2つの文字列型の列と10個のint型の列
- GCP Spannerインスタンス：5ノードのマルチリージョンnam6として構成
私の読み込みプログラムはGCPVMで実行され、Spannerインスタンスにアクセスする専用クライアントです。自動コミットが有効になっています。バッチ挿入は私のプログラムによって実行される唯一のDML操作であり、バッチサイズは約1500です。各コミットで、ミューテーション制限である20000を完全に使い果たします。同時に、コミットサイズは5MB（値2つの文字列型の列のうちの1つは小さいサイズです）。行は主キーの最初の列に基づいてパーティション化されるため、パフォーマンスを向上させるために、各コミットをごく少数のパーティションに送信できます。
上記のすべての構成と最適化により、挿入率は1秒あたり約1,000行になります。挿入する行が8億行を超えているため、これは本当に残念です。私は公式文書がおよそ言及していることに気づきました。マルチリージョンSpannerインスタンスのピーク書き込み（QPS合計）は1800です。
だから私はここに2つの質問があります：
- このような低いピーク書き込みQPSを考慮すると、GCPは、顧客が大規模なデータセットをマルチリージョンSpannerインスタンスに移行することを期待していない、またはサポートしていないことを意味しますか？
- Spannerの監視による読み取りレイテンシが高いことがわかりました。読み取り要求はありません。私の推測では、行を書き込んでいる間、Spannerは最初に読み取り、同じ主キーを持つ行が存在するかどうかを確認する必要があります。私の推測が正しければ、なぜそんなに時間がかかるのですか？そうでない場合、これらの読み取り操作がどのように行われるかについてのガイダンスを得ることができますか？
回答
データをロードするクライアントアプリケーションをどのように設定しているかは、私にはよくわかりません。私の最初の印象は、クライアントアプリケーションが十分なトランザクションを並行して実行していない可能性があるということです。通常、1,000行/秒を大幅に超える数を挿入できるはずですが、場合によっては複数のVMから複数のトランザクションを並行して実行する必要があります。次の簡単な例を使用して、ローカルマシンから単一ノードのSpannerインスタンスへの負荷スループットをテストしました。これにより、約1,500行/秒のスループットが得られました。
Spannerインスタンスと同じネットワークリージョン内の1つ以上のVMで実行されているクライアントアプリケーションを使用するマルチノードセットアップは、それよりも大きなボリュームを実現できるはずです。
import com.google.api.client.util.Base64;
import com.google.common.base.Stopwatch;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class TestJdbc {
public static void main(String[] args) {
final int threads = 512;
ExecutorService executor = Executors.newFixedThreadPool(threads);
watch = Stopwatch.createStarted();
for (int i = 0; i < threads; i++) {
executor.submit(new InsertRunnable());
}
}
static final AtomicLong rowCount = new AtomicLong();
static Stopwatch watch;
static final class InsertRunnable implements Runnable {
@Override
public void run() {
try (Connection connection =
DriverManager.getConnection(
"jdbc:cloudspanner:/projects/my-project/instances/my-instance/databases/my-db")) {
while (true) {
try (PreparedStatement ps =
connection.prepareStatement("INSERT INTO Test (Id, Col1, Col2) VALUES (?, ?, ?)")) {
for (int i = 0; i < 150; i++) {
ps.setLong(1, rnd.nextLong());
ps.setString(2, randomString(100));
ps.setString(3, randomString(100));
ps.addBatch();
rowCount.incrementAndGet();
}
ps.executeBatch();
}
System.out.println("Rows inserted: " + rowCount);
System.out.println("Rows/second: " + rowCount.get() / watch.elapsed(TimeUnit.SECONDS));
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
private final Random rnd = new Random();
private String randomString(int maxLength) {
byte[] bytes = new byte[rnd.nextInt(maxLength / 2) + 1];
rnd.nextBytes(bytes);
return Base64.encodeBase64String(bytes);
}
}
}
より良い結果を得るために調整を試みることができる他のいくつかのこともあります：
- バッチあたりの行数を減らすと、全体的な結果が向上する可能性があります。
- 可能であれば、
InsertOrUpdateミューテーションオブジェクトを使用すると、DMLステートメントを使用するよりもはるかに効率的です（以下の例を参照）。
MutationDMLの代わりに使用する例：
import com.google.api.client.util.Base64;
import com.google.cloud.spanner.Mutation;
import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
public class TestJdbc {
public static void main(String[] args) {
final int threads = 512;
ExecutorService executor = Executors.newFixedThreadPool(threads);
watch = Stopwatch.createStarted();
for (int i = 0; i < threads; i++) {
executor.submit(new InsertOrUpdateMutationRunnable());
}
}
static final AtomicLong rowCount = new AtomicLong();
static Stopwatch watch;
static final class InsertOrUpdateMutationRunnable implements Runnable {
@Override
public void run() {
try (Connection connection =
DriverManager.getConnection(
"jdbc:cloudspanner:/projects/my-project/instances/my-instance/databases/my-db")) {
CloudSpannerJdbcConnection csConnection = connection.unwrap(CloudSpannerJdbcConnection.class);
CloudSpannerJdbcConnection csConnection =
connection.unwrap(CloudSpannerJdbcConnection.class);
while (true) {
ImmutableList.Builder<Mutation> builder = ImmutableList.builder();
for (int i = 0; i < 150; i++) {
builder.add(
Mutation.newInsertOrUpdateBuilder("Test")
.set("Id")
.to(rnd.nextLong())
.set("Col1")
.to(randomString(100))
.set("Col2")
.to(randomString(100))
.build());
rowCount.incrementAndGet();
}
csConnection.write(builder.build());
System.out.println("Rows inserted: " + rowCount);
System.out.println("Rows/second: " + rowCount.get() / watch.elapsed(TimeUnit.SECONDS));
}
}
} catch (SQLException e) {
throw new RuntimeException(e);
}
}
private final Random rnd = new Random();
private String randomString(int maxLength) {
byte[] bytes = new byte[rnd.nextInt(maxLength / 2) + 1];
rnd.nextBytes(bytes);
return Base64.encodeBase64String(bytes);
}
}
}
上記の簡単な例では、さらに調整しなくても、約35,000行/秒のスループットが得られます。
追加情報2020-08-21：ミューテーションオブジェクトが（バッチ）DMLステートメントよりも効率的である理由は、DMLステートメントがCloud Spannerによって内部的に読み取りクエリに変換され、それがミューテーションの作成に使用されるためです。この変換は、バッチ内のすべてのDMLステートメントに対して実行する必要があります。つまり、1,500個の単純な挿入ステートメントを含むDMLバッチは、1,500個の（小さな）読み取りクエリをトリガーし、1,500個のミューテーションに変換する必要があります。これはおそらく、モニタリングで見られる読み取り遅延の背後にある理由でもあります。
それ以外の場合は、クライアントアプリケーションがどのように見えるか、および実行しているインスタンスの数に関する情報を共有していただけませんか。
挿入する行が8億を超え、Javaプログラマーであることがわかった場合、DataflowでBeamを使用することを提案できますか？
Beamのスパナライターは、書き込みを可能な限り効率的に行うように設計されています。つまり、行を同様のキーでグループ化し、実行中にバッチ処理します。Beam on Dataflowは、複数のワーカーVMを使用して、複数のファイル読み取りとスパナ書き込みを並行して実行することもできます。
マルチリージョンスパナインスタンスを使用すると、ノードあたり1秒あたり約1800行の挿入速度を得ることができ（Knutの回答が示唆するように、行が小さくバッチ処理されている場合はさらに多く）、5つのスパナノードを使用すると、おそらく10〜20行になります。並行して実行されるインポータースレッド-インポータープログラムを使用するか、Dataflowを使用するかに関係なく。
（開示：私はBeam SpannerIOのメンテナーです）