Transmitir alterações com o Dataflow

Com o conector Bigtable Beam, você usa o Dataflow para ler registros de alteração de dados do Bigtable sem precisar rastrear ou processar alterações de partição no seu código, porque o conector processa essa lógica para você.

Neste documento, descrevemos como configurar e usar o conector Bigtable Beam para ler um fluxo de alterações usando um pipeline do Dataflow. Antes de ler este documento, leia a Visão geral do fluxo de alterações e conheça o Dataflow.

Alternativas para criar seu próprio pipeline

Se você não quiser criar seu próprio canal do Dataflow, use uma das opções a seguir.

Você pode usar um modelo do Dataflow fornecido pelo Google.

Você também pode usar os exemplos de código do tutorial ou do guia de início rápido do Bigtable como ponto de partida para seu código.

Verifique se o código gerado usa o google cloud libraries-bom versão 26.14.0 ou posterior.

Detalhes do conector

Com o conector Bigtable Beam, chamado BigtableIO.readChangeStream, você lê um fluxo de registros de alteração de dados (ChangeStreamMutation) que podem ser processados. O conector do Bigtable Beam é um componente do repositório do Apache Beam no GitHub (em inglês). Para conferir uma descrição do código do conector, consulte os comentários em BigtableIO.java.

É preciso usar o conector com a versão 2.48.0 ou posterior do Beam. Verifique o suporte ao ambiente de execução do Apache Beam para garantir que você está usando uma versão compatível do Java. É possível implantar um pipeline que usa o conector no Dataflow, que lida com o provisionamento e o gerenciamento de recursos e ajuda na escalonabilidade e confiabilidade do processamento de dados do fluxo.

Para mais informações sobre o modelo de programação do Apache Beam, consulte a documentação do Beam.

Como agrupar dados sem horas de evento

Os registros de alteração de dados transmitidos usando o conector de Beam do Bigtable não são compatíveis com as funções do Dataflow que dependem de horários dos eventos.

Conforme explicado em Replicação e marcas-d'água, uma marca-d'água baixa poderá não avançar se a replicação da partição não tiver alcançado o restante da instância. Quando uma marca-d'água baixa deixa de avançar, isso pode interromper o fluxo de alterações.

Para evitar que o stream fique parado, o conector do Beam do Bigtable envia todos os dados com um carimbo de data/hora de saída igual a zero. O carimbo de data/hora zerado faz o Dataflow considerar todos os registros de alteração de dados como dados atrasados. Por isso, os recursos do Dataflow que dependem das horas de evento não são compatíveis com o fluxo de alterações do Bigtable. Especificamente, não é possível usar funções de gestão de janelas, gatilhos de hora de evento, ou timers de hora de evento.

Em vez disso, use GlobalWindows com gatilhos de horário sem evento para agrupar esses dados atrasados em painéis, conforme demonstrado em exemplo do tutorial. Para ver detalhes sobre gatilhos e painéis, consulte Gatilhos no guia de programação do Beam.

Escalonamento automático

O conector é compatível com o escalonamento automático do Dataflow, que é ativado por padrão ao usar o Runner portátil (obrigatório). O algoritmo do escalonamento automático do Dataflow considera o backlog estimado do fluxo de alterações, que pode ser monitorado na página Monitoramento do Dataflow na seção Backlog. Use a flag --maxNumWorkers ao implantar um job para limitar o número de workers.

Para escalonar manualmente o pipeline em vez de usar o escalonamento automático, consulte Como escalonar manualmente um pipeline de streaming.

Limitações

Observe as seguintes limitações antes de usar o conector Bigtable Beam com o Dataflow.

Dataflow Portable Runner

O conector só pode ser executado usando o Dataflow Portable Runner. Para ativá-lo, especifique --experiments=use_runner_v2 nos argumentos da linha de comando. A execução com o Streaming Java Runner causa falha no pipeline com a seguinte exceção:

java.lang.UnsupportedOperationException: BundleFinalizer unsupported by non-portable Dataflow

Snapshots

O conector não é compatível com snapshots do Dataflow.

Cópias

O conector Bigtable Beam transmite alterações para cada chave de linha e cada cluster na ordem do carimbo de data/hora de confirmação, mas, como às vezes é reiniciado de horários anteriores no stream, ele pode produzir duplicados.

Reinicializações de pipeline

Se um pipeline do Dataflow tiver sido interrompido por muito tempo, os registros de alteração de dados poderão ficar atrasados em relação ao limite de retenção. Quando o pipeline é retomado, o Bigtable falha para que você possa iniciar um novo pipeline com um novo horário de início da solicitação que esteja dentro do período de armazenamento. O Bigtable faz isso, em vez de avançar silenciosamente o horário da solicitação do pipeline original, para evitar a remoção não intencional de registros de alteração de dados com carimbos de data/hora que estão fora do período de armazenamento especificado.

Antes de começar

Antes de usar o conector, conclua os seguintes pré-requisitos.

Configurar a autenticação

Para usar os exemplos de Java nesta página em um ambiente de desenvolvimento local, instale e inicialize a CLI gcloud e configure o Application Default Credentials com suas credenciais de usuário.

  1. Instale a Google Cloud CLI.

  2. Ao usar um provedor de identidade (IdP) externo, primeiro faça login na CLI gcloud com sua identidade federada.

  3. Se você estiver usando um shell local, crie credenciais de autenticação local para sua conta de usuário: 

    gcloud auth application-default login

    Não é necessário fazer isso se você estiver usando o Cloud Shell.

    Se um erro de autenticação for retornado e você estiver usando um provedor de identidade (IdP) externo, confirme se você fez login na CLI gcloud com sua identidade federada.

Para mais informações, consulte Configurar a autenticação para um ambiente de desenvolvimento local.

Para informações sobre como configurar a autenticação para um ambiente de produção, consulte Configurar o Application Default Credentials para código em execução no Google Cloud .

Ativar um fluxo de alterações

Você precisa ativar um fluxo de alterações em uma tabela antes de poder lê-lo. Também é possível criar uma nova tabela com fluxo de alterações ativados.

Tabela de metadados do fluxo de alterações

Ao transmitir alterações com o Dataflow, o conector Bigtable Beam cria uma tabela de metadados chamada __change_stream_md_table por padrão. A tabela de metadados do fluxo de alterações gerencia o estado operacional do conector e armazena metadados sobre registros de alteração de dados.

Por padrão, o conector cria a tabela na mesma instância da tabela que está sendo transmitida. Para garantir que a tabela funcione corretamente, o perfil do aplicativo para a tabela de metadados precisa usar o roteamento de cluster único e ter transações de linha única ativadas.

Para mais informações sobre como transmitir alterações do Bigtable com o conector Bigtable Beam, consulte a documentação do BigtableIO.

Funções exigidas

Para receber as permissões necessárias para ler um fluxo de alterações do Bigtable usando o Dataflow, peça ao administrador para conceder a você os seguintes papéis do IAM.

Para ler as alterações do Bigtable, você precisa deste papel:

  • Administrador do Bigtable (roles/bigtable.admin) na instância do Bigtable que contém a tabela com as alterações que você pretende mostrar

Para executar o job do Dataflow, você precisa destes papéis:

Para mais informações sobre como conceder papéis, consulte Gerenciar acesso.

Também é possível conseguir as permissões necessárias com papéis personalizados ou outros papéis predefinidos.

Adicionar o conector do Bigtable Beam como dependência

Adicione um código semelhante à seguinte dependência ao arquivo pom.xml do Maven. A versão precisa ser 2.48.0 ou posterior.

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>VERSION</version>
  </dependency>
</dependencies>

Ler o fluxo de alterações

Para criar um pipeline do Dataflow para ler os registros de alteração de dados, configure o conector e adicione transformações e coletores. Em seguida, use o conector para ler objetos ChangeStreamMutation em um pipeline do Beam.

Os exemplos de código nesta seção, escritos em Java, demonstram como criar um pipeline e usá-lo para converter pares de chave-valor em uma string. Cada par consiste em uma chave de linha e um objeto ChangeStreamMutation. O pipeline converte as entradas de cada objeto em uma string separada por vírgulas.

Criar o pipeline

Este exemplo de código em Java demonstra como criar o pipeline:

BigtableOptions options =
    PipelineOptionsFactory.fromArgs(args).withValidation().as(BigtableOptions.class);
Pipeline p = Pipeline.create(options);

final Instant startTime = Instant.now();

p.apply(
        "Read Change Stream",
        BigtableIO.readChangeStream()
            .withProjectId(options.getBigtableProjectId())
            .withInstanceId(options.getBigtableInstanceId())
            .withTableId(options.getBigtableTableId())
            .withAppProfileId(options.getBigtableAppProfile())
            .withStartTime(startTime))
    .apply(
        "Flatten Mutation Entries",
        FlatMapElements.into(TypeDescriptors.strings())
            .via(ChangeStreamsHelloWorld::mutationEntriesToString))
    .apply(
        "Print mutations",
        ParDo.of(
            new DoFn<String, Void>() { // a DoFn as an anonymous inner class instance
              @ProcessElement
              public void processElement(@Element String mutation) {
                System.out.println("Change captured: " + mutation);
              }
            }));
p.run();

Processar os registros de alteração de dados

Neste exemplo, demonstramos como percorrer em repetição todas as entradas em um registro de alteração de dados de uma linha e chamar um método de conversão em string com base no tipo de entrada.

Veja uma lista dos tipos de entrada que um registro de alteração de dados pode conter em O que há em um registro de alteração de dados.

static List<String> mutationEntriesToString(KV<ByteString, ChangeStreamMutation> mutationPair) {
  List<String> mutations = new ArrayList<>();
  String rowKey = mutationPair.getKey().toStringUtf8();
  ChangeStreamMutation mutation = mutationPair.getValue();
  MutationType mutationType = mutation.getType();
  for (Entry entry : mutation.getEntries()) {
    if (entry instanceof SetCell) {
      mutations.add(setCellToString(rowKey, mutationType, (SetCell) entry));
    } else if (entry instanceof DeleteCells) {
      mutations.add(deleteCellsToString(rowKey, mutationType, (DeleteCells) entry));
    } else if (entry instanceof DeleteFamily) {
      // Note: DeleteRow mutations are mapped into one DeleteFamily per-family
      mutations.add(deleteFamilyToString(rowKey, mutationType, (DeleteFamily) entry));
    } else {
      throw new RuntimeException("Entry type not supported.");
    }
  }
  return mutations;
}

Neste exemplo, uma entrada de gravação é convertida:

private static String setCellToString(String rowKey, MutationType mutationType, SetCell setCell) {
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "SetCell",
          setCell.getFamilyName(),
          setCell.getQualifier().toStringUtf8(),
          setCell.getValue().toStringUtf8());
  return String.join(",", mutationParts);
}

Neste exemplo, uma entrada de exclusão de células é convertida:

private static String deleteCellsToString(
    String rowKey, MutationType mutationType, DeleteCells deleteCells) {
  String timestampRange =
      deleteCells.getTimestampRange().getStart() + "-" + deleteCells.getTimestampRange().getEnd();
  List<String> mutationParts =
      Arrays.asList(
          rowKey,
          mutationType.name(),
          "DeleteCells",
          deleteCells.getFamilyName(),
          deleteCells.getQualifier().toStringUtf8(),
          timestampRange);
  return String.join(",", mutationParts);
}

Neste exemplo, uma exclusão de um grupo de colunas é convertida:


private static String deleteFamilyToString(
    String rowKey, MutationType mutationType, DeleteFamily deleteFamily) {
  List<String> mutationParts =
      Arrays.asList(rowKey, mutationType.name(), "DeleteFamily", deleteFamily.getFamilyName());
  return String.join(",", mutationParts);
}

Monitoramento

Os recursos a seguir no Google Cloud console permitem monitorar seus Google Cloud recursos enquanto você executa um pipeline do Dataflow para ler um fluxo de alterações do Bigtable:

Verifique sobretudo as seguintes métricas:

  • Na página de insights do sistema do Bigtable, verifique as seguintes métricas:
    • Dados de utilização da CPU por fluxo de alterações na métrica cpu_load_by_app_profile_by_method_by_table. Mostra o impacto do fluxo de alterações no uso da CPU do cluster.
    • Utilização do armazenamento do fluxo de alterações (bytes) (change_stream_log_used_bytes).

  • Na página de monitoramento do Dataflow, verifique a atualização de dados. Essa métrica mostra a diferença entre a hora atual e a marca-d'água, que é de aproximadamente dois minutos, com picos ocasionais de um ou dois minutos a mais. A atualização de dados não indica se os registros de alteração de dados estão sendo processados lentamente. Para garantir a integridade e o desempenho contínuos dos aplicativos críticos, monitore a métrica de atualização de dados do Dataflow e siga estas etapas:

    • Se a métrica de atualização de dados for consistentemente maior que o limite, o pipeline poderá ter poucos recursos. Recomendamos que você adicione mais workers do Dataflow.
    • Se os workers do Dataflow estiverem bem provisionados, mas a atualização de dados tiver aumentado ou estiver consistentemente alta, entre em contato com o Google Cloud suporte.

  • A métrica processing_delay_from_commit_timestamp_MEAN do Dataflow pode informar o tempo médio de processamento de registros de alteração de dados durante o ciclo de vida do job.

A métrica server/latencies do Bigtable não é útil ao monitorar um pipeline do Dataflow que está lendo um fluxo de alterações do Bigtable, porque reflete a duração da solicitação de streaming, não a latência de processamento do registro de alteração de dados. A alta latência em um fluxo de alterações não significa que as solicitações estão sendo processadas lentamente. Isso significa que a conexão ficou aberta por muito tempo.

A seguir