Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3
Nesta página, você verá como criar uma arquitetura de push baseada em eventos acionando DAGs do Cloud Composer em resposta a alterações nos tópicos do Pub/Sub. Os exemplos neste tutorial demonstram como lidar com o ciclo completo de gerenciamento do Pub/Sub, incluindo o gerenciamento de assinaturas, como parte do processo do DAG. Ele é adequado para alguns dos casos de uso comuns em que você precisa acionar DAGs, mas não quer configurar permissões de acesso extras.
Por exemplo, as mensagens enviadas pelo Pub/Sub podem ser usadas como uma solução se você não quiser fornecer acesso direto a um ambiente do Cloud Composer por motivos de segurança. É possível configurar uma função do Cloud que cria mensagens do Pub/Sub e as publica em um tópico do Pub/Sub. Em seguida, crie um DAG que extraia mensagens do Pub/Sub e as processe.
Neste exemplo específico, você cria uma função do Cloud e implanta dois DAGs. O primeiro DAG extrai mensagens do Pub/Sub e aciona o segundo de acordo com o conteúdo da mensagem do Pub/Sub.
Para acompanhar este tutorial, é necessário ter familiaridade com o Python e o console do Google Cloud.
Objetivos
Custos
Neste tutorial, usamos o seguinte componente faturável do Google Cloud:
- Cloud Composer (consulte também os custos adicionais)
- Pub/Sub
- Cloud Functions
Depois de concluir este tutorial, exclua os recursos criados para evitar o faturamento contínuo. Consulte Limpeza para mais detalhes.
Antes de começar
Para este tutorial, você precisa de um projeto do Google Cloud. Configure o projeto da seguinte maneira:
No console do Google Cloud, selecione ou crie um projeto:
Verifique se o faturamento foi ativado para o projeto. Saiba como verificar se o faturamento está ativado em um projeto.
Verifique se o usuário do projeto do Google Cloud tem os seguintes papéis para criar os recursos necessários:
- Usuário da conta de serviço (
roles/iam.serviceAccountUser
) - Editor do Pub/Sub (
roles/pubsub.editor
) - Administrador de ambiente e de objetos do Storage
(
roles/composer.environmentAndStorageObjectAdmin
) - Administrador do Cloud Functions (
roles/cloudfunctions.admin
) - Visualizador de registros (
roles/logging.viewer
)
- Usuário da conta de serviço (
Verifique se a conta de serviço que executa a função do Cloud tem permissões suficientes no projeto para acessar o Pub/Sub. Por padrão, o Cloud Functions usa a conta de serviço padrão do App Engine. Essa conta de serviço tem o papel de Editor, com permissões suficientes para este tutorial.
Ativar as APIs do projeto
Console
Ative as APIs Cloud Composer, Cloud Functions, and Pub/Sub.
gcloud
Ative as APIs Cloud Composer, Cloud Functions, and Pub/Sub:
gcloud services enable composer.googleapis.comcloudfunctions.googleapis.com pubsub.googleapis.com
Terraform
Ative a API Cloud Composer no seu projeto adicionando as seguintes definições de recursos ao script do Terraform:
resource "google_project_service" "composer_api" {
project = "<PROJECT_ID>"
service = "composer.googleapis.com"
// Disabling Cloud Composer API might irreversibly break all other
// environments in your project.
// This parameter prevents automatic disabling
// of the API when the resource is destroyed.
// We recommend to disable the API only after all environments are deleted.
disable_on_destroy = false
}
resource "google_project_service" "pubsub_api" {
project = "<PROJECT_ID>"
service = "pubsub.googleapis.com"
disable_on_destroy = false
}
resource "google_project_service" "functions_api" {
project = "<PROJECT_ID>"
service = "cloudfunctions.googleapis.com"
disable_on_destroy = false
}
Substitua <PROJECT_ID>
pelo ID do projeto. Por exemplo, example-project
.
criar o ambiente do Cloud Composer
Crie um ambiente do Cloud Composer 2.
Como parte desse procedimento,
você concede o papel Extensão do agente de serviço da API Cloud Composer v2
(roles/composer.ServiceAgentV2Ext
) à conta do agente de serviço do
Composer. O Cloud Composer usa essa conta para executar operações
no projeto do Google Cloud.
Criar um tópico do Pub/Sub.
Este exemplo aciona um DAG em resposta a uma mensagem enviada para um tópico do Pub/Sub. Crie um tópico do Pub/Sub para usar neste exemplo:
Console
No console do Google Cloud, acesse a página Tópicos do Pub/Sub.
Clique em Criar tópico.
No campo ID do tópico, insira
dag-topic-trigger
como o ID do tópico.Não altere as outras opções.
Clique em Criar tópico.
gcloud
Para criar um tópico, execute o comando gcloud pubsub topics create na Google Cloud CLI:
gcloud pubsub topics create dag-topic-trigger
Terraform
Adicione as seguintes definições de recursos ao script do Terraform:
resource "google_pubsub_topic" "trigger" {
project = "<PROJECT_ID>"
name = "dag-topic-trigger"
message_retention_duration = "86600s"
}
Substitua <PROJECT_ID>
pelo ID do projeto. Por exemplo, example-project
.
Fazer upload dos DAGs
Faça upload dos DAGs para seu ambiente:
- Salve o seguinte arquivo DAG no seu computador local.
- Substitua
<PROJECT_ID>
pelo ID do projeto. Por exemplo,example-project
. - Faça upload do arquivo DAG editado para seu ambiente.
O exemplo de código contém dois DAGs: trigger_dag
e target_dag
.
O DAG trigger_dag
se inscreve em um tópico do Pub/Sub, extrai
mensagens do Pub/Sub e aciona outro DAG especificado no ID do DAG
dos dados das mensagens do Pub/Sub. Neste exemplo, trigger_dag
aciona o DAG target_dag
, que envia mensagens para os registros de tarefas.
O DAG trigger_dag
contém as seguintes tarefas:
subscribe_task
: assina um tópico do Pub/Sub.pull_messages_operator
: lê os dados de uma mensagem do Pub/Sub comPubSubPullOperator
.trigger_target_dag
: aciona outro DAG (neste exemplo,target_dag
) de acordo com os dados nas mensagens extraídas do tópico do Pub/Sub.
O DAG target_dag
contém apenas uma tarefa: output_to_logs
. Ela
imprime mensagens no registro de tarefas com um segundo de atraso.
Implantar uma função do Cloud que publique mensagens em um tópico do Pub/Sub
Nesta seção, você implantará uma função do Cloud que publica mensagens em um tópico do Pub/Sub.
Criar uma função do Cloud e especificar a configuração dela
Console
No Console do Google Cloud, acesse a página Cloud Functions.
Clique em Criar função.
No campo Ambiente selecione 1a geração.
No campo Nome da função, insira o nome da sua função:
pubsub-publisher
.No campo Tipo de gatilho, selecione HTTP.
Na seção Autenticação, selecione Permitir invocações não autenticadas. Essa opção concede a usuários não autenticados a capacidade de invocar uma função HTTP.
Clique em "Salvar".
Clique em Próximo para acessar a etapa Código.
Terraform
Considere usar o console do Google Cloud nesta etapa, porque não há uma maneira simples de gerenciar o código-fonte da função no Terraform.
Este exemplo demonstra como fazer upload de uma função do Cloud de um arquivo ZIP local criando um bucket do Cloud Storage, armazenando o arquivo nesse bucket e usando o arquivo do bucket como fonte para a função do Cloud. Se você usar essa abordagem, o Terraform não atualizará automaticamente o código-fonte da sua função, mesmo que você crie um novo arquivo. Para fazer um novo upload do código da função, mude o nome do arquivo.
- Carregue os arquivos
pubsub_publisher.py
erequirements.txt
. - No arquivo
pubsub_publisher.py
, substitua<PROJECT_ID>
pelo ID do projeto. Por exemplo,example-project
. - Crie um arquivo ZIP chamado
pubsub_function.zip
com os arquivospbusub_publisner.py
erequirements.txt
. - Salve o arquivo ZIP no diretório em que o script do Terraform está armazenado.
- Adicione as seguintes definições de recurso ao script do Terraform e substitua
<PROJECT_ID>
pelo ID do seu projeto.
resource "google_storage_bucket" "cloud_function_bucket" {
project = <PROJECT_ID>
name = "<PROJECT_ID>-cloud-function-source-code"
location = "US"
force_destroy = true
uniform_bucket_level_access = true
}
resource "google_storage_bucket_object" "cloud_function_source" {
name = "pubsub_function.zip"
bucket = google_storage_bucket.cloud_function_bucket.name
source = "./pubsub_function.zip"
}
resource "google_cloudfunctions_function" "pubsub_function" {
project = <PROJECT_ID>
name = "pubsub-publisher"
runtime = "python310"
region = "us-central1"
available_memory_mb = 128
source_archive_bucket = google_storage_bucket.cloud_function_bucket.name
source_archive_object = "pubsub_function.zip"
timeout = 60
entry_point = "pubsub_publisher"
trigger_http = true
}
Especificar parâmetros do código da função do Cloud
Console
Na etapa Código, no campo Ambiente de execução, selecione a linguagem que a função usa. Neste exemplo, selecione Python 3.10.
No campo Ponto de entrada, insira
pubsub_publisher
. Esse é o código executado quando a função do Cloud é executada. O valor dessa sinalização precisa ser um nome de função ou um nome de classe totalmente qualificado que exista no código-fonte.
Terraform
Ignore esta etapa. Os parâmetros da função do Cloud já estão definidos no
recurso google_cloudfunctions_function
.
Fazer upload do código da função do Cloud
Console
No campo Código-fonte, selecione a opção apropriada para como você fornece o código-fonte da função. Neste tutorial, adicione o código da função usando o Editor in-line do Cloud Functions. Como alternativa, você pode fazer upload de um arquivo ZIP ou usar o Cloud Source Repositories.
- Coloque o exemplo de código a seguir no arquivo main.py.
- Substitua
<PROJECT_ID>
pelo ID do projeto. Por exemplo,example-project
.
Terraform
Ignore esta etapa. Os parâmetros da função do Cloud já estão definidos no
recurso google_cloudfunctions_function
.
Especificar as dependências da função do Cloud
Console
Especifique as dependências da função no arquivo de metadados requirements.txt:
Quando você implanta a função, o Cloud Functions faz o download e instala as dependências declaradas no arquivo requirements.txt, uma linha por pacote.
Esse arquivo precisa estar no mesmo diretório do arquivo main.py que contém o código da função. Para saber mais, consulte Arquivos de requisitos na documentação do pip
.
Terraform
Ignore esta etapa. As dependências da função do Cloud são definidas no
arquivo requirements.txt
no arquivo pubsub_function.zip
.
Implantar a função do Cloud
Console
Selecione Implantar. Quando a implantação for concluída com êxito, a função aparecerá com uma marca de seleção verde na página do Cloud Functions no console do Google Cloud.
Verifique se a conta de serviço que executa a função do Cloud tem permissões suficientes no projeto para acessar o Pub/Sub.
Terraform
Inicialize o Terraform:
terraform init
Revise a configuração e verifique se os recursos que o Terraform vai criar ou atualizar atendem às suas expectativas:
terraform plan
Para verificar se a configuração é válida, execute o seguinte comando:
terraform validate
Aplique a configuração do Terraform executando o seguinte comando e inserindo "yes" no prompt:
terraform apply
Aguarde até que o Terraform exiba a mensagem "Apply complete!".
No console do Google Cloud, navegue até os recursos na UI para verificar se o Terraform os criou ou atualizou.
Testar a Função do Cloud
Para verificar se a função publica uma mensagem em um tópico do Pub/Sub e se os DAGs de exemplo funcionam conforme o esperado:
Verifique se os DAGs estão ativos:
No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia DAGs.
Verifique os valores na coluna Estado para DAGs chamados
trigger_dag
etarget_dag
. Os dois DAGs precisam estar no estadoActive
.
Envie uma mensagem de teste do Pub/Sub. Faça isso no Cloud Shell:
No console do Google Cloud, acesse a página Funções.
Clique no nome da função,
pubsub-publisher
.Acesse a guia Testes.
Na seção Configurar evento de acionamento, insira a seguinte chave-valor JSON:
{"message": "target_dag"}
. Não modifique o par de chave-valor, porque essa mensagem aciona o DAG de teste mais tarde.Na seção Testar comando, clique em Testar no Cloud Shell.
No Terminal do Cloud Shell, aguarde até que um comando apareça automaticamente. Execute este comando pressionando
Enter
.Se a mensagem Autorizar o Cloud Shell for exibida, clique em Autorizar.
Verifique se o conteúdo da mensagem corresponde à mensagem do Pub/Sub. Neste exemplo, a mensagem de saída precisa começar com
Message b'target_dag' with message_length 10 published to
como uma resposta da sua função.
Verifique se
target_dag
foi acionado:Aguarde pelo menos um minuto para que uma nova execução do DAG de
trigger_dag
seja concluída.No console do Google Cloud, acesse a página Ambientes.
Na lista de ambientes, clique no nome do ambiente. A página Detalhes do ambiente é aberta.
Acesse a guia DAGs.
Clique em
trigger_dag
para acessar a página Detalhes do DAG. Na guia Execuções, é exibida uma lista de execuções do DAGtrigger_dag
.Esse DAG é executado a cada minuto e processa todas as mensagens do Pub/Sub enviadas da função. Se nenhuma mensagem tiver sido enviada, a tarefa
trigger_target
será marcada comoSkipped
nos registros de execução do DAG. Se os DAGs foram acionados, a tarefatrigger_target
será marcada comoSuccess
.Analise várias execuções recentes de DAGs para localizar uma execução em que todas as três tarefas (
subscribe_task
,pull_messages_operator
etrigger_target
) estejam em statusSuccess
.Volte para a guia DAGs e verifique se a coluna Execuções bem-sucedidas do DAG
target_dag
lista uma execução bem-sucedida.
Resumo
Neste tutorial, você aprendeu a usar o Cloud Functions para publicar mensagens em um tópico do Pub/Sub e implantar um DAG que se inscreve em um tópico do Pub/Sub, extrai mensagens do Pub/Sub e aciona outro DAG especificado no ID do DAG dos dados da mensagem.
Há também formas alternativas de criar e gerenciar assinaturas do Pub/Sub e acionar DAGs que estão fora do escopo deste tutorial. Por exemplo, é possível usar o Cloud Functions para acionar DAGs do Airflow quando ocorrer um evento especificado. Confira nossos tutoriais para testar os outros recursos do Google Cloud.
Limpar
Para evitar cobranças na sua conta do Google Cloud pelos recursos usados neste tutorial, exclua o projeto que os contém ou mantenha o projeto e exclua os recursos individuais.
Exclua o projeto
Exclua um projeto do Google Cloud:
gcloud projects delete PROJECT_ID
Excluir recursos individuais
Se você planeja ver vários tutoriais e guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do projeto.
Console
- Exclua o ambiente do Cloud Composer. Você também vai excluir o bucket do ambiente durante este procedimento.
- Exclua o tópico do Pub/Sub,
dag-topic-trigger
. Exclua a função do Cloud.
No Console do Google Cloud, acesse o Cloud Functions.
Clique na caixa de seleção da função que você quer excluir,
pubsub-publisher
.Clique em Excluir e siga as instruções.
Terraform
- Verifique se o script do Terraform não contém entradas para recursos que ainda são necessários para seu projeto. Por exemplo, é possível manter algumas APIs ativadas e permissões do IAM ainda atribuídas (se você tiver adicionado essas definições ao script do Terraform).
- Execute
terraform destroy
. - Exclua manualmente o bucket do ambiente. O Cloud Composer não a exclui automaticamente. É possível fazer isso no console ou na Google Cloud CLI.
A seguir
- Como testar DAGs
- Testar funções HTTP
- Implantar uma função do Cloud
- Teste outros recursos do Google Cloud. Confira nossos tutoriais.