jdbc kullanarak Spanner'a satır eklerken düşük yükleme performansı
Arka plan: TSV biçimli veri dosyalarını (MySQL veritabanından dökülmüş) bir GCP Spanner tablosuna yüklemeye çalışıyorum.
- istemci kitaplığı: resmi Spanner JDBC bağımlılığı v1.15.0
- tablo şeması: iki dize tipli sütun ve on dahili tipli sütun
- GCP Spanner örneği: 5 düğümlü çok bölgeli nam6 olarak yapılandırıldı
Yükleme programım GCP sanal makinesinde çalışıyor ve Spanner örneğine erişen özel istemcidir. Otomatik kesinleştirme etkinleştirildi. Toplu ekleme, programım tarafından yürütülen tek DML işlemidir ve toplu iş boyutu 1500 civarındadır. Her işlemede, 20000 olan mutasyon sınırını tam olarak kullanır. Aynı zamanda, kesinleştirme boyutu 5 MB'nin altındadır (değerler dizge türü iki sütun küçük boyutludur). Satırlar, birincil anahtarın ilk sütununa göre bölümlenir, böylece her işlem daha iyi performans için çok az bölüme gönderilebilir.
Yukarıdaki tüm yapılandırma ve optimizasyonla, ekleme oranı saniyede yalnızca yaklaşık 1 bin satırdır. Bu beni gerçekten hayal kırıklığına uğratıyor çünkü eklemek için 800 milyondan fazla satırım var. Resmi belgenin yaklaşık olarak bahsettiğini fark ettim . en yüksek yazma (QPS toplamı), çok bölgeli Spanner örneği için 1800'dür.
Yani burada iki sorum var:
- Bu kadar düşük yazma QPS değeri düşünüldüğünde, GCP'nin müşterilerin büyük veri kümelerini çok bölgeli Spanner örneğine taşımasını beklemediği veya desteklemediği anlamına mı geliyor?
- Spanner izlemeden gelen yüksek okuma gecikmesini görüyordum. Herhangi bir okuma talebim yok. Tahminimce, satırları yazarken Spanner'ın önce aynı birincil anahtara sahip bir satırın var olup olmadığını okuması ve kontrol etmesi gerekiyor. Tahminim doğruysa, neden bu kadar zaman alıyor? Değilse, bu okuma işlemlerinin nasıl gerçekleştiği konusunda herhangi bir rehberlik alabilir miyim?
Yanıtlar
Verileri yükleyen istemci uygulamasını nasıl kurduğunuz benim için tam olarak net değil. İlk izlenimim, müşteri uygulamanızın paralel olarak yeterince işlem yürütmediği yönünde. Normalde saniyede 1.000 satırdan önemli ölçüde daha fazlasını ekleyebilmeniz gerekir, ancak bu, muhtemelen birden çok VM'den birden çok işlemi paralel olarak yürütmenizi gerektirir. Yerel makinemden tek bir Spanner örneğine yük verimini test etmek için aşağıdaki basit örneği kullandım ve bu bana yaklaşık 1.500 satır / saniye iş hacmi sağladı.
Spanner örneğinizle aynı ağ bölgesinde bir veya daha fazla sanal makinede çalışan bir istemci uygulamasını kullanan çok düğümlü bir kurulum, bundan daha yüksek hacimler elde edebilmelidir.
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);
}
}
}
Daha iyi sonuçlar almak için ayarlamaya çalışabileceğiniz birkaç başka şey daha var:
- Parti başına satır sayısını azaltmak, daha iyi genel sonuçlar verebilir.
- Mümkünse,
InsertOrUpdatemutasyon nesnelerini kullanmak DML ifadelerini kullanmaktan çok daha etkilidir (aşağıdaki örneğe bakın).
MutationDML yerine kullanım örneği :
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);
}
}
}
Yukarıdaki basit örnek, daha fazla ayarlama yapmadan bana yaklaşık 35.000 satır / saniye iş hacmi veriyor.
EK BİLGİ 2020-08-21 : Mutasyon nesnelerinin (toplu) DML ifadelerinden daha verimli olmasının nedeni, DML ifadelerinin dahili olarak Cloud Spanner tarafından okuma sorgularına dönüştürülerek daha sonra mutasyonlar oluşturmak için kullanılmasıdır. Bu dönüştürme, bir toplu işteki her DML ifadesi için yapılmalıdır; bu, 1.500 basit insert deyimine sahip bir DML grubunun, 1.500 (küçük) okuma sorgusunu tetikleyeceği ve 1.500 mutasyona dönüştürülmesi gerektiği anlamına gelir. Bu, büyük olasılıkla, izlemenizde gördüğünüz okuma gecikmesinin arkasındaki nedendir.
İstemci uygulamanızın neye benzediği ve kaç tane örneğini çalıştırdığınız hakkında daha fazla bilgi paylaşır mısınız?
Eklenecek 800 milyondan fazla satırla ve bir Java programcısı olduğunuzu görünce, Dataflow'da Beam kullanmayı önerebilir miyim?
Kiriş spanner yazar benzer anahtar ile satırları gruplandırma ve yapıyorsun olarak dozajları - onun yazma ile mümkün olduğunca verimli olacak şekilde tasarlanmıştır. Beam on Dataflow, birden çok dosya okuma ve anahtar yazımını paralel olarak yürütmek için birkaç çalışan sanal makinesi de kullanabilir ...
Çok bölgeli bir anahtar örneğiyle, saniyede düğüm başına yaklaşık 1800 satır ekleme hızı (Knut'un yanıtının önerdiği gibi satırlar küçük ve topluysa daha fazla) ve 5 anahtar düğümü ile, muhtemelen 10 ila 20 arasında olabilir. ithalatçı programınızı veya Dataflow'u kullanarak paralel çalışan içe aktarıcı iş parçacıkları.
(açıklama: Ben Beam SpannerIO bakımcısıyım)