Nesta página, você aprenderá a criar um pipeline de streaming usando como exemplo um modelo do Dataflow fornecido pelo Google. Mais especificamente, o modelo do Pub/Sub para BigQuery é usado como exemplo.
Antes de começar
-
Faça login na sua conta do Google.
Se você ainda não tiver uma, inscreva-se.
-
No Console do Google Cloud, na página do seletor de projetos, selecione ou crie um projeto do Google Cloud.
-
Verifique se o faturamento está ativado para seu projeto na nuvem. Saiba como confirmar se o faturamento está ativado para o projeto.
- Ative as APIs Cloud Dataflow, Compute Engine, Stackdriver Logging, Cloud Storage, Cloud Storage JSON, BigQuery, Cloud Pub/Sub e Cloud Resource Manager .
- Crie um bucket do Cloud Storage:
- No Console do Cloud, acesse a página Navegador do Cloud Storage.
- Clique em Criar bucket.
- Na caixa de diálogo Criar bucket, especifique os seguintes atributos.
- Nome: um nome de bucket exclusivo. Não inclua informações confidenciais no nome do bucket: o namespace é global e visível ao público.
- Classe de armazenamento padrão: Standard
- Um local onde os dados do bucket serão armazenados.
- Clique em Criar.
Crie um tópico
- Acesse a página de tópicos do Pub/Sub na IU da Web.
Acesse a página de tópicos do Pub/Sub - Clique em addCriar tópico.
- No campo ID do tópico, forneça um nome de tópico exclusivo,
por exemplo,
taxirides-realtime
. - Clique em Save.
Criar um conjunto de dados e uma tabela do BigQuery
Crie um conjunto de dados do BigQuery e uma tabela com o esquema apropriado para seu tópico Pub/Sub, usando o Cloud Shell ou o Console do Cloud.
Neste exemplo, o nome do conjunto de dados é taxirides
e o nome da tabela é realtime
.
Como usar o Cloud Shell
Use o Cloud Shell para criar um conjunto de dados e uma tabela.
- Crie o conjunto de dados executando o comando:
bq mk taxirides
A saída será semelhante a esta:Dataset “myprojectid:taxirides” successfully created
- Crie a tabela executando o comando:
bq mk \ --time_partitioning_field timestamp \ --schema ride_id:string,point_idx:integer,latitude:float,longitude:float,\ timestamp:timestamp,meter_reading:float,meter_increment:float,ride_status:string,\ passenger_count:integer -t taxirides.realtime
A saída será semelhante a esta:Table “myprojectid:taxirides.realtime” successfully created
A tabela é particionada para reduzir os custos de consulta e melhorar o desempenho.
Como usar o Console do Google Cloud Platform
Use o Console do Google Cloud para criar um conjunto de dados e uma tabela.
- Acesse a IU da Web do BigQuery.
ACESSAR A IU DA WEB DO BIGQUERY - Clique no ícone de seta para baixo ao lado do nome do projeto e selecione Criar conjunto de dados. Insira
taxirides
como o ID do conjunto de dados.Os IDs de conjunto de dados são exclusivos por projeto. Clique no ícone de ponto de interrogação para ver as limitações do código.
- Deixe as outras configurações padrão como estão e clique em OK.
- Na navegação, mantenha o ponteiro do mouse sobre o ID do conjunto de dados que você acabou de criar. Clique no ícone de seta para baixo ao lado do código e depois clique em Criar nova tabela.
- Ao lado de Dados de origem, selecione a opção Criar tabela vazia.
- Em Tabela de destino, selecione
taxirides
e digiterealtime
. - Em Esquema, selecione Editar como texto e insira:
ride_id:string,point_idx:integer,latitude:float,longitude:float,timestamp:timestamp, meter_reading:float,meter_increment:float,ride_status:string,passenger_count:integer
- Em Opções, selecione a opção Dia no campo Tipo de particionamento.
- Em Opções, selecione a coluna de carimbo de data/hora no seletor do campo Tipo de particionamento.
- Clique no botão Criar tabela.

Executar o canal
Execute um pipeline de streaming usando o modelo de tópico do Pub/Sub para BigQuery fornecido pelo Google.
- Acesse a IU da Web do Dataflow.
ACESSAR A IU DA WEB DO CLOUD DATAFLOW - Clique em Criar job usando um modelo.
- Digite um Nome do job para o job do Dataflow.
- Em modelo do Dataflow, selecione o modelo tópico do Pub/Sub para BigQuery.
- Em Tópico de entrada do Pub/Sub, digite
projects/pubsub-public-data/topics/taxirides-realtime
. O pipeline recebe dados de entrada a partir do tópico de entrada. - Em Tabela de saída do BigQuery, digite
myprojectid:taxirides.realtime
. - Em Local temporário, digite
gs://mybucket/temp/
. Essa subpasta serve para armazenar arquivos temporários, como o job preparado do pipeline. - Clique no botão Executar job.
- Veja os dados gravados no BigQuery. Acesse a IU da Web do BigQuery.
ACESSAR A IU DA WEB DO BIGQUERY
É possível enviar consultas usando o SQL padrão. Por exemplo, a consulta a seguir seleciona todas as linhas que foram adicionadas nas últimas 24 horas:SELECT * FROM `myprojectid.taxirides.realtime` WHERE `timestamp` > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY) LIMIT 1000

Limpar
Para evitar cobranças na sua conta do Google Cloud pelos recursos usados neste guia de início rápido, siga estas etapas:
- Acesse a IU da Web do Dataflow.
ACESSAR A IU DA WEB DO CLOUD DATAFLOW - Acesse a IU da Web do BigQuery.
ACESSAR A IU DA WEB DO BIGQUERY- No painel de navegação, passe o cursor sobre o conjunto de dados taxirides criado.
- Clique no ícone da seta para baixo ao lado do nome do conjunto de dados na navegação e depois clique em Excluir conjunto de dados.
- Na caixa de diálogo Excluir conjunto de dados, confirme a exclusão digitando o nome do conjunto de dados ("taxirides") e clicando em OK.