Use o Datastream e o Dataflow para fazer stream de dados para o BigQuery

Nesta página, encontra práticas recomendadas para usar o Datastream e o Dataflow para fazer stream de dados para o BigQuery.

Particione conjuntos de dados de réplicas em chaves definidas pelo utilizador

O conjunto de dados de preparação no BigQuery é particionado automaticamente. No entanto, por predefinição, o conjunto de dados da réplica não é particionado porque as chaves de partição nas tabelas da réplica têm de ser definidas com base numa lógica empresarial específica, em vez de serem aplicadas pelo Datastream e Dataflow.

Para cada tabela no conjunto de dados da réplica que precisa de partição:

  1. Pare e esvazie a tarefa do Dataflow

  2. Use o editor de SQL no BigQuery para executar o seguinte script SQL para cada tabela no conjunto de dados da réplica. Para este exemplo, a tabela actor no conjunto de dados datastream_cdc tem uma coluna last_update que queremos definir como a nossa chave de partição. Ao executar o script, recria a tabela com a chave de partição correta.

    create table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' partition by date(last_update)
as SELECT * FROM '[BigQuery_PROJECT_ID].datastream_cdc.actor'

drop table '[BigQuery_PROJECT_ID].datastream_cdc.actor'

alter table '[BigQuery_PROJECT_ID].datastream_cdc.actor_new' rename to 'actor'

  3. Use o modelo Datastream para BigQuery para recriar uma tarefa do Dataflow.

Executar funções definidas pelo utilizador para manipular dados de eventos

Pode usar o modelo Datastream para o BigQuery para executar uma função definida pelo utilizador em Javascript. Para o fazer, primeiro, coloque um ficheiro que contenha a função numa localização específica no Cloud Storage. Em seguida, faça o seguinte:

  • Use o parâmetro javascriptTextTransformGcsPath no modelo para especificar a localização do ficheiro no Cloud Storage que contém a função definida pelo utilizador.
  • Use o parâmetro javascriptTextTransformFunctionName para especificar o nome da função Javascript que quer chamar como função definida pelo utilizador.

Por exemplo, pode executar uma função definida pelo utilizador para reter registos eliminados nas tabelas do conjunto de dados da réplica no BigQuery. Este processo é conhecido como eliminação temporária.

Para tal, crie uma função que copie o valor da coluna _metadata_deleted para uma nova coluna denominada is_deleted e, em seguida, repõe o valor da coluna _metadata_deleted para false. Isto faz com que a tarefa do Dataflow ignore os eventos de eliminação e retenha os registos eliminados quando atualiza o conjunto de dados da réplica no BigQuery.

Segue-se o exemplo de código para esta função definida pelo utilizador:

/**
* This function changes the behavior of the Datastream to
* BigQuery template to allow soft deletes.
* @param {string} messageString from DatastreamIO data
* @return {string} same as an input message with an added property
*/
function transform(messageString) {
   // messageString is a JSON object as a string
   var messageJson = JSON.parse(messageString);
    // Moving the deleted flag to a new column will cause the pipeline to soft delete data.
   messageJson['is_deleted'] = messageJson['_metadata_deleted'];
   messageJson['_metadata_deleted'] = false;
    return JSON.stringify(messageJson);
 }

Defina a frequência de união

Use o parâmetro mergeFrequencyMinutes do modelo do Datastream para o BigQuery para definir a frequência de união. Este é o número de minutos entre as uniões de uma determinada tabela no conjunto de dados da réplica no BigQuery. Enquanto os dados do histórico são preenchidos novamente, recomendamos que mantenha a frequência de união baixa (12 ou 24 horas) para manter os custos sob controlo.

Por exemplo, se definir o valor deste parâmetro como 10 minutos, o Dataflow executa a tarefa que usa o modelo a cada 10 minutos. No entanto, na primeira vez que a tarefa é executada, existe um atraso de 5 minutos. Neste exemplo, se a tarefa for executada às 09:14, a primeira união ocorre às 09:29 (10 minutos para a união e 5 minutos para o atraso). A segunda união ocorre às 09:39 e todas as uniões subsequentes ocorrem em intervalos de 10 minutos (09:49, 09:59, 10:09 e assim sucessivamente).

Se definir a frequência de união como 60 minutos, a tarefa é executada de hora a hora, após um atraso de 5 minutos para a execução inicial da tarefa. Se a tarefa estiver agendada para ser executada às 10:00, é executada às 10:05 devido ao atraso de 5 minutos. Todas as sincronizações subsequentes ocorrem em intervalos de 60 minutos (11:05, 12:05, 13:05, etc.).

Quer seja devido ao controlo dos custos ou por outros motivos, pode não conseguir fazer uma união com uma frequência que satisfaça as necessidades da sua empresa. Pode não ter os dados mais recentes. Para aceder aos dados mais recentes, crie uma vista sobre as tabelas dos conjuntos de dados de preparação e réplica no BigQuery, em que a vista imita a união. Esta vista é criada como uma tabela lógica (para os conjuntos de dados de preparação e de réplica). Se a frequência de união for baixa e precisar de acesso mais rápido aos dados, use a vista.