baixo desempenho de carregamento ao inserir linhas em lote no Spanner usando jdbc

Question

Contexto: Estou tentando carregar arquivos de dados formatados em TSV (despejados do banco de dados MySQL) em uma tabela do GCP Spanner.

• biblioteca cliente: a dependência oficial do Spanner JDBC v1.15.0
• esquema da tabela: duas colunas do tipo string e dez colunas do tipo int
• Instância do GCP Spanner: configurado como nam6 multirregional com 5 nós

Meu programa de carregamento é executado no GCP VM e é o cliente exclusivo que acessa a instância do Spanner. A confirmação automática está ativada. A inserção em lote é a única operação DML executada pelo meu programa e o tamanho do lote é em torno de 1500. Em cada commit, ele usa totalmente o limite de mutação, que é 20.000. E ao mesmo tempo, o tamanho do commit está abaixo de 5 MB (os valores de duas colunas do tipo string são de tamanho pequeno). As linhas são particionadas com base na primeira coluna da chave primária para que cada confirmação possa ser enviada para muito poucas partições para melhor desempenho.

Com toda a configuração e otimização acima, a taxa de inserção é de apenas cerca de 1 mil linhas por segundo. Isso realmente me decepciona porque tenho mais de 800 milhões de linhas para inserir. Eu percebi que o médico oficial mencionou o aprox. O pico de gravação (QPS total) é 1800 para a instância do Spanner multirregional.

Portanto, tenho duas perguntas aqui:

1. Considerando esse QPS de gravação de baixo pico, isso significa que o GCP não espera ou não oferece suporte aos clientes para migrar grandes conjuntos de dados para a instância do Spanner multirregional?
2. Eu estava vendo a alta latência de leitura do monitoramento do Spanner. Eu não tenho nenhum pedido de leitura. Meu palpite é que, ao escrever linhas, o Spanner precisa primeiro ler e verificar se existe uma linha com a mesma chave primária. Se meu palpite estiver certo, por que leva tanto tempo? Se não, posso obter alguma orientação sobre como essas operações de leitura acontecem?
   
   

Answer 1

Não está muito claro para mim exatamente como você está configurando o aplicativo cliente que está carregando os dados. Minha impressão inicial é que seu aplicativo cliente pode não estar executando transações suficientes em paralelo. Normalmente, você deve ser capaz de inserir significativamente mais de 1.000 linhas / segundo, mas isso exigiria que você execute várias transações em paralelo, possivelmente de várias VMs. Usei o seguinte exemplo simples para testar a taxa de transferência de carga de minha máquina local para uma instância do Spanner de nó único , e isso me deu uma taxa de transferência de aproximadamente 1.500 linhas / segundo.

Uma configuração de vários nós usando um aplicativo cliente em execução em uma ou mais VMs na mesma região de rede que sua instância do Spanner deve ser capaz de atingir volumes maiores do que isso.

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

Existem também algumas outras coisas que você pode tentar ajustar para obter melhores resultados:

• Reduzir o número de linhas por lote pode produzir melhores resultados gerais.
• Se possível, usar InsertOrUpdateobjetos de mutação é muito mais eficiente do que usar instruções DML (veja o exemplo abaixo).

Exemplo usando em Mutationvez de DML:

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

O exemplo simples acima me dá uma taxa de transferência de aproximadamente 35.000 linhas / segundo sem qualquer ajuste adicional.

INFORMAÇÕES ADICIONAIS 21-08-2020 : O motivo pelo qual os objetos de mutação são mais eficientes do que as instruções DML (em lote) é que as instruções DML são convertidas internamente em consultas de leitura pelo Cloud Spanner, que são então usadas para criar mutações. Essa conversão precisa ser feita para cada instrução DML em um lote, o que significa que um lote DML com 1.500 instruções de inserção simples acionará 1.500 (pequenas) consultas de leitura e precisará ser convertido em 1.500 mutações. Provavelmente, essa também é a razão por trás da latência de leitura que você está vendo em seu monitoramento.

Caso contrário, você se importaria de compartilhar mais algumas informações sobre a aparência de seu aplicativo cliente e quantas instâncias dele você está executando?

Answer 2

Com mais de 800 milhões de linhas para inserir e visto que você é um programador Java, posso sugerir o uso do Beam no Dataflow?

O gravador de chaves no Beam foi projetado para ser o mais eficiente possível com suas gravações - agrupando linhas por uma chave semelhante e agrupando-as em lote conforme você está fazendo. O Beam no Dataflow também pode usar várias VMs de trabalho para executar várias leituras de arquivos e gravações de chave em paralelo ...

Com uma instância de chave inglesa multirregional, você deve ser capaz de obter aproximadamente 1800 linhas por nó por segundo de velocidade de inserção (mais se as linhas forem pequenas e em lote, como sugere a resposta de Knut) e com 5 nós de chave inglesa, você provavelmente pode ter entre 10 e 20 threads do importador em execução em paralelo - seja usando seu programa de importação ou usando Dataflow.

(divulgação: eu sou o mantenedor do Beam SpannerIO)