Modelo de streams de alterações do Bigtable para o BigQuery

O modelo de streams de alterações do Bigtable para o BigQuery é um pipeline de streaming que faz stream de registos de alterações de dados do Bigtable e escreve-os em tabelas do BigQuery através do Dataflow.

Uma stream de alterações do Bigtable permite-lhe subscrever mutações de dados por tabela. Quando subscreve streams de alterações de tabelas, aplicam-se as seguintes restrições:

  • Apenas são devolvidas as células modificadas e os descritores das operações de eliminação.
  • Apenas é devolvido o novo valor de uma célula modificada.

Quando os registos de alterações de dados são escritos no BigQuery, as linhas podem ser inseridas fora de ordem em comparação com a ordenação da data/hora de confirmação original do Bigtable.

As linhas da tabela do registo de alterações que não podem ser escritas no BigQuery devido a um erro persistente são colocadas permanentemente num diretório de fila de mensagens rejeitadas (fila de mensagens não processadas) no Cloud Storage para revisão humana ou processamento adicional pelo utilizador.

Se a tabela do BigQuery necessária não existir, o pipeline cria-a. Caso contrário, é usada uma tabela do BigQuery existente. O esquema das tabelas do BigQuery existentes tem de conter as colunas na tabela seguinte.

Cada nova linha do BigQuery inclui um registo de alteração de dados devolvido pela stream de alterações da respetiva linha na tabela do Bigtable.

Esquema da tabela de saída do BigQuery

Nome da coluna Tipo Permitido valor nulo Descrição
row_key STRING ou BYTES Não A chave da linha alterada. Quando a opção de pipeline writeRowkeyAsBytes está definida como true, o tipo da coluna tem de ser BYTES. Caso contrário, use o tipo STRING.
mod_type STRING Não O tipo de mutação de linhas. Use um dos seguintes valores: SET_CELL, DELETE_CELLS ou DELETE_FAMILY.
column_family STRING Não A família de colunas afetada pela mutação de linhas.
column STRING Sim O qualificador da coluna afetado pela mutação da linha. Para o tipo de mutação DELETE_FAMILY, defina como NULL.
commit_timestamp TIMESTAMP Não A hora em que o Bigtable aplica a mutação.
big_query_commit_timestamp TIMESTAMP Sim Opcional: especifica a hora em que o BigQuery escreve a linha numa tabela de saída. O campo não é preenchido se o nome da coluna estiver presente no valor da opção da conduta bigQueryChangelogTableFieldsToIgnore.
timestamp TIMESTAMP ou INT64 Sim O valor da data/hora da célula afetada pela mutação. Quando a opção de pipeline writeNumericTimestamps está definida como true, o tipo da coluna tem de ser INT64. Caso contrário, use o tipo TIMESTAMP. Para os tipos de mutação DELETE_CELLS e DELETE_FAMILY, defina como NULL.
timestamp_from TIMESTAMP ou INT64 Sim Descreve um início inclusivo do intervalo de indicações de tempo para todas as células eliminadas pela mutação DELETE_CELLS. Para outros tipos de mutações, defina como NULL.
timestamp_to TIMESTAMP ou INT64 Sim Descreve um fim exclusivo do intervalo de data/hora para todas as células eliminadas pela mutação DELETE_CELLS. Para outros tipos de mutações, defina como NULL.
is_gc BOOL Não Opcional: quando a mutação é acionada por uma política de recolha de lixo, defina como true. Em todos os outros casos, defina como false. O campo não é preenchido quando o nome da coluna está presente no valor da opção da pipeline bigQueryChangelogTableFieldsToIgnore.
source_instance STRING Não Opcional: descreve o nome da instância do Bigtable de onde provém a mutação. O campo não é preenchido quando o nome da coluna está presente no valor da opção da pipeline bigQueryChangelogTableFieldsToIgnore.
source_cluster STRING Não Opcional: descreve o nome do cluster do Bigtable de onde provém a mutação. O campo não é preenchido quando o nome da coluna está presente no valor da opção da pipeline bigQueryChangelogTableFieldsToIgnore.
source_table STRING Não Opcional: descreve o nome da tabela do Bigtable à qual a mutação se aplica. O valor nesta coluna pode ser útil se várias tabelas do Bigtable transmitirem alterações para a mesma tabela do BigQuery. O campo não é preenchido quando o nome da coluna está presente no valor da opção da pipeline bigQueryChangelogTableFieldsToIgnore.
tiebreaker INT64 Não Opcional: quando duas mutações são registadas ao mesmo tempo por diferentes clusters do Bigtable, a mutação com o valor tiebreaker mais elevado é aplicada à tabela de origem. As mutações com valores de tiebreaker mais baixos são rejeitadas. O campo não é preenchido quando o nome da coluna está presente no valor da opção da pipeline bigQueryChangelogTableFieldsToIgnore.
value STRING ou BYTES Sim O novo valor definido pela mutação. Quando a opção de pipeline writeValuesAsBytes está definida como true, o tipo da coluna tem de ser BYTES. Caso contrário, use o tipo STRING. O valor é definido para mutações de SET_CELL. Para outros tipos de mutações, o valor é definido como NULL.

Requisitos do pipeline

  • A instância de origem do Bigtable especificada.
  • A tabela de origem do Bigtable especificada. A tabela tem de ter streams de alterações ativadas.
  • O perfil da aplicação do Bigtable especificado.
  • O conjunto de dados de destino do BigQuery especificado.

Parâmetros de modelos

Parâmetros obrigatórios

  • bigQueryDataset: o nome do conjunto de dados da tabela do BigQuery de destino.
  • bigtableChangeStreamAppProfile: o ID do perfil da aplicação do Bigtable. O perfil da aplicação tem de usar o encaminhamento de cluster único e permitir transações de linha única.
  • bigtableReadInstanceId: o ID da instância do Bigtable de origem.
  • bigtableReadTableId: o ID da tabela do Bigtable de origem.

Parâmetros opcionais

  • writeRowkeyAsBytes: indica se os rowkeys devem ser escritos como BYTES do BigQuery. Quando definido como true, as chaves de linhas são escritas na coluna BYTES. Caso contrário, os rowkeys são escritos na coluna STRING. A predefinição é false.
  • writeValuesAsBytes: quando definido como true, os valores são escritos numa coluna do tipo BYTES. Caso contrário, são escritos numa coluna do tipo STRING. Predefinição: false.
  • writeNumericTimestamps: indica se a data/hora do Bigtable deve ser escrita como INT64 do BigQuery. Quando definido como true, os valores são escritos na coluna INT64. Caso contrário, os valores são escritos na coluna TIMESTAMP. Colunas afetadas: timestamp, timestamp_from e timestamp_to. A predefinição é false. Quando definido como true, o tempo é medido em microssegundos desde a época Unix (1 de janeiro de 1970 às 00:00:00 UTC).
  • bigQueryProjectId: o ID do projeto do conjunto de dados do BigQuery. A predefinição é o projeto da tarefa do Dataflow.
  • bigQueryChangelogTableName: nome da tabela de registo de alterações do BigQuery de destino. Se não for especificado, é usado o valor bigtableReadTableId + "_changelog". A predefinição é vazio.
  • bigQueryChangelogTablePartitionGranularity: especifica um nível de detalhe para a partição da tabela do histórico de alterações. Quando definido, a tabela é particionada. Use um dos seguintes valores suportados: HOUR, DAY, MONTH ou YEAR. Por predefinição, a tabela não é particionada.
  • bigQueryChangelogTablePartitionExpirationMs: define o tempo de validade da partição da tabela do registo de alterações, em milissegundos. Quando definido como true, as partições com uma antiguidade superior ao número de milissegundos especificado são eliminadas. Por predefinição, não é definida nenhuma data de validade.
  • bigQueryChangelogTableFieldsToIgnore: uma lista separada por vírgulas das colunas do histórico de alterações que, quando especificadas, não são criadas nem preenchidas. Use um dos seguintes valores suportados: is_gc, source_instance, source_cluster, source_table, tiebreaker ou big_query_commit_timestamp. Por predefinição, todas as colunas são preenchidas.
  • dlqDirectory: o diretório a usar para a fila de mensagens rejeitadas. Os registos que não são processados são armazenados neste diretório. A predefinição é um diretório na localização temporária da tarefa do Dataflow. Na maioria dos casos, pode usar o caminho predefinido.
  • bigtableChangeStreamMetadataInstanceId: o ID da instância de metadados das streams de alterações do Bigtable. A predefinição é vazio.
  • bigtableChangeStreamMetadataTableTableId: o ID da tabela de metadados do conetor de streams de alterações do Bigtable. Se não for fornecida, é criada automaticamente uma tabela de metadados do conector de streams de alterações do Bigtable durante a execução do pipeline. A predefinição é vazio.
  • bigtableChangeStreamCharset: o nome do conjunto de carateres das streams de alterações do Bigtable. A predefinição é: UTF-8.
  • bigtableChangeStreamStartTimestamp: a data/hora de início (https://tools.ietf.org/html/rfc3339), inclusive, a usar para ler streams de alterações. Por exemplo, 2022-05-05T07:59:59Z. O valor predefinido é a data/hora de início do pipeline.
  • bigtableChangeStreamIgnoreColumnFamilies: uma lista separada por vírgulas de alterações de nomes de famílias de colunas a ignorar. A predefinição é vazio.
  • bigtableChangeStreamIgnoreColumns: uma lista separada por vírgulas de alterações de nomes de colunas a ignorar. Exemplo: "cf1:col1,cf2:col2". A predefinição é vazio.
  • bigtableChangeStreamName: um nome exclusivo para o pipeline do cliente. Permite-lhe retomar o processamento a partir do ponto em que um pipeline em execução anterior parou. A predefinição é um nome gerado automaticamente. Consulte os registos da tarefa do Dataflow para ver o valor usado.
  • bigtableChangeStreamResume: quando definido como true, um novo pipeline retoma o processamento a partir do ponto em que um pipeline executado anteriormente com o mesmo valor bigtableChangeStreamName parou. Se o pipeline com o valor bigtableChangeStreamName especificado nunca tiver sido executado, não é iniciado um novo pipeline. Quando definido como false, é iniciado um novo pipeline. Se já tiver sido executado um pipeline com o mesmo valor de bigtableChangeStreamName para a origem especificada, não é iniciado um novo pipeline. A predefinição é false.
  • bigtableReadChangeStreamTimeoutMs: o limite de tempo para pedidos Bigtable ReadChangeStream em milissegundos.
  • bigtableReadProjectId: o ID do projeto do Bigtable. A predefinição é o projeto da tarefa do Dataflow.

Execute o modelo

Consola

  1. Aceda à página Dataflow Criar tarefa a partir de um modelo.
    2. Aceda a Criar tarefa a partir de modelo
  2. No campo Nome da tarefa, introduza um nome exclusivo para a tarefa.
  3. Opcional: para Ponto final regional, selecione um valor no menu pendente. A região predefinida é us-central1.

    Para ver uma lista das regiões onde pode executar uma tarefa do Dataflow, consulte o artigo Localizações do Dataflow.

  4. No menu pendente Modelo do fluxo de dados, selecione the Bigtable change streams to BigQuery template.
  5. Nos campos de parâmetros fornecidos, introduza os valores dos parâmetros.
  6. Clique em Executar tarefa.

gcloud

Na shell ou no terminal, execute o modelo:

gcloud dataflow flex-template run JOB_NAME \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery \
    --parameters \
bigtableReadInstanceId=BIGTABLE_INSTANCE_ID,\
bigtableReadTableId=BIGTABLE_TABLE_ID,\
bigtableChangeStreamAppProfile=BIGTABLE_APPLICATION_PROFILE_ID,\
bigQueryDataset=BIGQUERY_DESTINATION_DATASET

Substitua o seguinte:

  • PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
  • JOB_NAME: um nome de tarefa exclusivo à sua escolha
  • VERSION: a versão do modelo que quer usar

    Pode usar os seguintes valores:

  • REGION_NAME: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
  • BIGTABLE_INSTANCE_ID: o ID da instância do Bigtable.
  • BIGTABLE_TABLE_ID: o ID da tabela do Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: o ID do perfil da aplicação do Bigtable.
  • BIGQUERY_DESTINATION_DATASET: o nome do conjunto de dados de destino do BigQuery

API

Para executar o modelo através da API REST, envie um pedido HTTP POST. Para mais informações sobre a API e os respetivos âmbitos de autorização, consulte projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
  "launch_parameter": {
    "jobName": "JOB_NAME",
    "containerSpecGcsPath": "gs://dataflow-templates-REGION_NAME/VERSION/flex/Bigtable_Change_Streams_to_BigQuery",
    "parameters": {
        "bigtableReadInstanceId": "BIGTABLE_INSTANCE_ID",
        "bigtableReadTableId": "BIGTABLE_TABLE_ID",
        "bigtableChangeStreamAppProfile": "BIGTABLE_APPLICATION_PROFILE_ID",
        "bigQueryDataset": "BIGQUERY_DESTINATION_DATASET"
    }
  }
}

Substitua o seguinte:

  • PROJECT_ID: o ID do projeto onde quer executar a tarefa do Dataflow Google Cloud
  • JOB_NAME: um nome de tarefa exclusivo à sua escolha
  • VERSION: a versão do modelo que quer usar

    Pode usar os seguintes valores:

  • LOCATION: a região onde quer implementar a tarefa do Dataflow, por exemplo, us-central1
  • BIGTABLE_INSTANCE_ID: o ID da instância do Bigtable.
  • BIGTABLE_TABLE_ID: o ID da tabela do Bigtable.
  • BIGTABLE_APPLICATION_PROFILE_ID: o ID do perfil da aplicação do Bigtable.
  • BIGQUERY_DESTINATION_DATASET: o nome do conjunto de dados de destino do BigQuery

O que se segue?