Neste tutorial, mostramos como implantar um pipeline de dados no Dataflow para um fluxo em tempo real das alterações do banco de dados provenientes de um fluxo de alterações da tabela do Bigtable. A saída do pipeline é gravada em uma série de arquivos no Cloud Storage.
Um exemplo de conjunto de dados para um aplicativo de música é fornecido. Neste tutorial, você rastreia as músicas ouvidas e, em seguida, classifica as cinco principais ao longo de um período.
Este tutorial é destinado a usuários técnicos familiarizados com a escrita de código e a implantação de pipelines de dados no Google Cloud.
Prepare o ambiente
Buscar o código
Clone o repositório GitHub que contém o código de exemplo: Se você já fez o download desse repositório, extraia a versão mais recente.
git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
cd java-docs-samples/bigtable/beam/change-streams
Criar um bucket
gcloud storage buckets create gs://BUCKET_NAME
BUCKET_NAME
with a bucket name
that meets the bucket naming requirements.
Criar uma instância do Bigtable
É possível usar uma instância deste tutorial ou criar uma instância com as configurações padrão em uma região perto de você.
Criar uma tabela
O aplicativo de amostra rastreia as músicas que os usuários ouvem e armazena os eventos de detecção no Bigtable. Crie uma tabela com um fluxo de alterações ativado que tenha um grupo de colunas (cf) e uma coluna (música) e que use IDs de usuário para chaves de linha.
Crie a tabela.
gcloud bigtable instances tables create song-rank \
--column-families=cf --change-stream-retention-period=7d \
--instance=BIGTABLE_INSTANCE_ID --project=PROJECT_ID
Substitua:
- PROJECT_ID: o ID do projeto que você está usando.
- BIGTABLE_INSTANCE_ID: o ID da instância para conter a nova tabela.
Iniciar o pipeline
Esse pipeline transforma o stream de alteração fazendo o seguinte:
- Lê o fluxo de alterações
- Obtém o nome da música
- Agrupa os eventos de música para ouvir em janelas de N segundos
- Conta as cinco músicas principais
- Gera os resultados
Executar o pipeline.
mvn compile exec:java -Dexec.mainClass=SongRank \
"-Dexec.args=--project=PROJECT_ID --bigtableProjectId=PROJECT_ID \
--bigtableInstanceId=BIGTABLE_INSTANCE_ID --bigtableTableId=song-rank \
--outputLocation=gs://BUCKET_NAME/ \
--runner=dataflow --region=BIGTABLE_REGION --experiments=use_runner_v2"
Substitua BIGTABLE_REGION pelo ID da região em que sua instância do Bigtable está, como us-east5
.
Entender o pipeline
Os snippets de código do pipeline a seguir podem ajudar a entender o código que você está executando.
Como ler o fluxo de alterações
O código nesta amostra configura o fluxo de origem com os parâmetros para a instância e a tabela específicas do Bigtable.
Como conseguir o nome da música
Quando uma música é ouvida, o nome dela é gravado no grupo de colunas cf
e no qualificador de coluna song
. Assim, o código extrai o valor da mutação do fluxo de alterações e o envia para a próxima etapa do pipeline.
Como contar as cinco músicas principais
É possível usar as funções Count
e Top.of
integradas do Beam para encontrar facilmente as cinco músicas principais na janela atual.
Como gerar os resultados
Esse pipeline grava os resultados na saída padrão e nos arquivos. Para os arquivos, ele exibe as gravações em grupos de 10 elementos ou segmentos de um minuto.
Ver o pipeline
No Google Cloud console, acesse a página Dataflow.
Clique no job com um nome que comece com song-rank.
Na parte inferior da tela, clique em Mostrar para abrir o painel de registros.
Clique em Registros do worker para monitorar os registros de saída do fluxo de alterações.
Fazer streaming de gravações
Use a CLI cbt
para gravar várias faixas de música para vários usuários na tabela song-rank
. Ela é projetada para gravar durante alguns minutos e simular o streaming de detecções de músicas ao longo do tempo.
cbt -instance=BIGTABLE_INSTANCE_ID -project=PROJECT_ID import \
song-rank song-rank-data.csv column-family=cf batch-size=1
Veja o resultado
Leia a saída no Cloud Storage para ver as músicas mais famosas.
gcloud storage cat gs://BUCKET_NAME/song-charts/GlobalWindow-pane-0-00000-of-00001.txt
Exemplo de saída:
2023-07-06T19:53:38.232Z [KV{The Wheels on the Bus, 199}, KV{Twinkle, Twinkle, Little Star, 199}, KV{Ode to Joy , 192}, KV{Row, Row, Row Your Boat, 186}, KV{Take Me Out to the Ball Game, 182}]
2023-07-06T19:53:49.536Z [KV{Old MacDonald Had a Farm, 20}, KV{Take Me Out to the Ball Game, 18}, KV{Für Elise, 17}, KV{Ode to Joy , 15}, KV{Mary Had a Little Lamb, 12}]
2023-07-06T19:53:50.425Z [KV{Twinkle, Twinkle, Little Star, 20}, KV{The Wheels on the Bus, 17}, KV{Row, Row, Row Your Boat, 13}, KV{Happy Birthday to You, 12}, KV{Over the Rainbow, 9}]