Criar um conector de origem do Cloud SQL para PostgreSQL

Neste documento, descrevemos como criar um conector de origem do Cloud SQL para PostgreSQL para o Kafka Connect.

Um conector de origem do Cloud SQL para PostgreSQL é uma instância de um conector do Debezium PostgreSQL. Ele lê mudanças no nível da linha de um banco de dados do Cloud SQL para PostgreSQL e as grava em tópicos em um cluster do Serviço Gerenciado para Apache Kafka.

Casos de uso desse conector:

  • Monitore mudanças no banco de dados no nível da linha em tempo real.
  • Integrar eventos de mudança de banco de dados a uma arquitetura orientada a eventos.
  • Responder a eventos do banco de dados, como inserções ou exclusões de linhas.
  • Copiar mudanças no banco de dados para outros sistemas.

Antes de começar

Antes de criar um conector de origem do Cloud SQL para PostgreSQL, verifique se você tem o seguinte:

Papéis e permissões necessárias

Para receber as permissões necessárias para criar um conector, peça ao administrador para conceder a você o papel do IAM de Editor do conector gerenciado do Kafka (roles/managedkafka.connectorEditor) no projeto. Para mais informações sobre a concessão de papéis, consulte Gerenciar o acesso a projetos, pastas e organizações.

Esse papel predefinido contém as permissões necessárias para criar um conector. Para acessar as permissões exatas necessárias, expanda a seção Permissões necessárias:

Permissões necessárias

As seguintes permissões são necessárias para criar um conector:

  • Crie um conector: managedkafka.connectors.create

Essas permissões também podem ser concedidas com funções personalizadas ou outros papéis predefinidos.

Conceder permissões para leitura do Cloud SQL

A conta de serviço do Managed Kafka precisa ter permissão para acessar o Cloud SQL para PostgreSQL. Conceda os seguintes papéis do IAM à conta de serviço:

  • Cliente do Cloud SQL (roles/cloudsql.client)
  • Usuário da instância do Cloud SQL (roles/cloudsql.instanceUser)

A conta de serviço do Managed Kafka tem o seguinte formato: service-PROJECT_NUMBER@gcp-sa-managedkafka., em que PROJECT_NUMBER é o número do projeto do cluster do Connect.

Se o cluster do Connect estiver em um projeto diferente do cluster do Serviço Gerenciado para Apache Kafka, consulte Criar um cluster do Connect em um projeto diferente.

Configurar o banco de dados

Antes de criar o conector, configure a replicação do banco de dados e permita que o conector faça a autenticação com o banco de dados. As seções a seguir descrevem essas etapas.

Ativar decodificação lógica

Um conector de origem do Cloud SQL para PostgreSQL depende do recurso de decodificação lógica do PostgreSQL. Para ativar a decodificação lógica na sua instância do Cloud SQL para PostgreSQL, siga estas etapas.

Console

  1. Acesse Cloud SQL > Instâncias.

    Acesse "Instâncias"

  2. Clique no nome da instância.

  3. Clique em Editar.

  4. Abra Sinalizações e parâmetros.

  5. Clique em Adicionar um flag de banco de dados.

  6. Na lista Escolha um sinalizador, selecione cloudsql.logical_decoding.

  7. Em Valor, selecione On.

  8. Clique em Concluído.

  9. Clique em Salvar.

Para mais informações, consulte Configurar a replicação lógica e a decodificação.

Configurar a captura de dados alterados (CDC)

Depois de ativar a decodificação lógica na sua instância, ative a captura de dados alterados (CDC) para as tabelas que você quer replicar.

Para ativar a CDC em uma tabela, execute a instrução SQL CREATE PUBLICATION. Essa instrução cria uma publicação, que define um grupo de tabelas a serem replicadas.

  • Opção 1. Crie uma publicação que replique as mudanças em todas as tabelas do banco de dados.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Opção 2. Crie uma publicação para um conjunto específico de tabelas.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    Substitua TABLE_LIST por uma lista separada por vírgulas de tabelas, no formato "schema_name"."table_name". Colocar os nomes do esquema e da tabela entre aspas duplas, como mostrado, evita erros de sintaxe se os nomes contiverem caracteres especiais ou letras maiúsculas.

Por padrão, o conector usa dbz_publication para o nome da publicação. Para usar uma publicação com um nome diferente, consulte Nome da publicação.

Criar uma conta de usuário para a conta de serviço do Kafka gerenciado

O conector de origem do Cloud SQL para PostgreSQL usa a autenticação de banco de dados do IAM para se conectar ao banco de dados. Para ativar a autenticação do banco de dados do IAM, adicione a conta de serviço do Kafka gerenciado à instância do Cloud SQL da seguinte maneira:

Console

  1. Acesse Cloud SQL > Instâncias

    Acesse "Instâncias"

  2. Clique no nome da instância.

  3. No painel de navegação, clique em Usuários.

  4. Clique em Adicionar conta de usuário.

  5. No painel Adicionar uma conta de usuário, selecione Cloud IAM.

  6. No campo Principal do IAM, insira o seguinte:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    Substitua PROJECT_NUMBER pelo número do projeto do cluster do Connect.

  7. Clique em Adicionar.

gcloud

Execute o comando gcloud sql users create:

gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
  --instance=INSTANCE_NAME \
  --type=cloud_iam_service_account

Substitua:

  • PROJECT_NUMBER: o número do projeto do cluster do Connect.

  • INSTANCE_NAME: o nome da instância do Cloud SQL para PostgreSQL.

Devido ao limite de tamanho do nome de usuário do banco de dados, o sufixo . é removido do nome de usuário, que passa a ser service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. Ao executar consultas SQL que fazem referência à conta de usuário do IAM, especifique o nome truncado.

Configurar a conta de usuário

Depois de criar a conta de usuário do IAM, conecte-se ao banco de dados como um usuário com o papel cloudsqlsuperuser (como o usuário padrão postgres ) e execute as seguintes consultas SQL.

Console

  1. Permite que o usuário leia o registro prévio de gravação.

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

  2. Conceda ao usuário a permissão SELECT nas tabelas.

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME"
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

    Como alternativa, é possível conceder a permissão SELECT em tabelas individuais. Se você escolher essa opção, também precisará definir a propriedade de configuração table.include.list do conector como a lista de tabelas permitidas. A consulta SQL a seguir concede a permissão SELECT em uma única tabela:

    GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

  3. Para cada tabela, conceda ao usuário acesso ao esquema dela. É possível pular esta etapa se a tabela estiver no esquema padrão public.

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

Configurar rede

Um conector de origem do Cloud SQL para PostgreSQL pode se conectar à instância do Cloud SQL das seguintes maneiras:

  • IP particular
  • Private Service Connect
  • IP público

Para mais informações sobre essas opções, consulte Escolher como se conectar ao Cloud SQL. Como prática recomendada de segurança, é recomendável usar o IP privado ou o Private Service Connect, porque essas opções não exigem conexão com um endereço IP externo.

A tabela a seguir mostra os requisitos de rede para cada opção:

Tipo de endereço IP Requisitos
IP particular Configure o IP particular para sua instância. Para mais informações, consulte Configurar IP privado.
Private Service Connect
  1. Configure o Private Service Connect para sua instância e receba o nome DNS do endpoint do Private Service Connect. Para mais informações, consulte Conectar a uma instância usando o Private Service Connect.
  2. Adicione o nome DNS do endpoint aos Domínios DNS resolvíveis do cluster do Connect. Para mais informações, consulte Atualizar um cluster do Connect.
IP público
  1. Configure o IP público da instância. Para mais informações, consulte Configurar o IP público.
  2. Configure o Public NAT para permitir que os workers do cluster do Connect se comuniquem com a Internet. Para mais informações, consulte Configurar o Public NAT. Ao criar o gateway do Cloud NAT, especifique a rede VPC que contém a sub-rede principal do cluster do Connect.

Criar um conector de origem do Cloud SQL para PostgreSQL

Para criar um conector de origem do Cloud SQL para PostgreSQL, siga estas etapas.

Quando o conector é inicializado, ele executa as seguintes ações:

  1. Cria um snapshot inicial do banco de dados.
  2. Cria um tópico do Kafka para cada tabela que tem linhas.
  3. Para cada linha do banco de dados, envia um evento de mudança ao tópico correspondente.

Enquanto o conector está em execução, ele continua enviando eventos de mudança para os tópicos. Para mais informações sobre o snapshot inicial, consulte Snapshots na documentação do Debezium.

Console

  1. No console do Google Cloud , acesse a página Conectar clusters.

    Acessar o Connect Clusters

  2. Clique no cluster do Connect em que você quer criar o conector.

  3. Clique em Criar conector.

  4. Para o nome do conector, insira uma string.

    Para conferir as diretrizes de nomeação de conectores, acesse Diretrizes de nomeação de recursos do Serviço gerenciado para Apache Kafka.

  5. Em Plug-in do conector, selecione Origem do Cloud SQL para PostgreSQL.

  6. Na lista Instância, selecione a instância do Cloud SQL.

  7. Na lista Banco de dados, selecione o banco de dados do Cloud SQL.

  8. No campo Prefixo do tópico, insira um prefixo para usar nos nomes dos tópicos do Kafka. Escolha um prefixo exclusivo para cada conector de origem do Cloud SQL para PostgreSQL.

  9. Opcional: no campo Nomes das tabelas, insira uma lista separada por vírgulas de tabelas para ler os dados de mudança, no formato "schema_name"."table_name". Se você deixar esse campo em branco, o conector vai ler os dados de mudança de todas as tabelas não relacionadas ao sistema no banco de dados.

  10. Opcional: na caixa Configurações, adicione propriedades de configuração ou edite as propriedades padrão. Para mais informações, consulte Configurar o conector.

    Talvez seja necessário substituir os padrões das seguintes propriedades:

  11. Opcional: selecione a Política de reinicialização da tarefa. Para mais informações, consulte Política de reinicialização de tarefas.

  12. Clique em Criar.

gcloud

  1. No console do Google Cloud , ative o Cloud Shell.

    Ativar o Cloud Shell

    Na parte de baixo do console Google Cloud , uma sessão do Cloud Shell é iniciada e exibe um prompt de linha de comando. O Cloud Shell é um ambiente shell com a CLI do Google Cloud já instalada e com valores já definidos para o projeto atual. A inicialização da sessão pode levar alguns segundos.

  2. Execute o comando gcloud managed-kafka connectors create:

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    Substitua:

    • CONNECTOR_ID: o ID ou nome do conector. Para conferir as diretrizes de nomeação de conectores, acesse Diretrizes de nomeação de recursos do Serviço gerenciado para Apache Kafka. O nome de um conector é imutável.

    • LOCATION: o local em que você cria o conector. Precisa ser o mesmo local em que você criou o cluster do Connect.

    • CONNECT_CLUSTER_ID: o ID do cluster do Connect em que o conector é criado.

    • CONFIG_FILE: o caminho para o arquivo de configuração YAML do conector.

    Confira um exemplo de arquivo de configuração para o conector de origem do Cloud SQL para PostgreSQL:

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
driver.cloudSqlInstance: INSTANCE_ID
driver.enableIamAuth: "true"
driver.ipTypes: IP_TYPES
driver.sslmode: disable
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
slot.name: SLOT_NAME
table.include.list: TABLE_LIST
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Substitua:

    • INSTANCE_ID: o ID da instância do Cloud SQL que contém o banco de dados, formatado da seguinte maneira:

      PROJECT_ID:REGION:INSTANCE_NAME

    • DATABASE_NAME: o nome do banco de dados do Cloud SQL de onde ler.

    • IP_TYPES: uma lista separada por vírgulas de tipos de endereços IP

    • SLOT_NAME: o nome do slot de replicação a ser criado.

    • TABLE_LIST: uma lista separada por vírgulas de tabelas para ler dados de mudança no formato "schema_name"."table_name".

    • TOPIC_PREFIX: um prefixo a ser usado para os nomes de tópicos do Kafka.

Configurar o conector

Nesta seção, descrevemos algumas propriedades de configuração que podem ser definidas no conector. Para uma lista completa, consulte Conector do Debezium para PostgreSQL na documentação do Debezium.

Tipos de endereço IP

A propriedade driver.ipTypes especifica o tipo de endereço IP usado pelo conector para se conectar ao banco de dados:

  • PRIVATE: IP privado
  • PSC: Private Service Connect
  • PUBLIC: IP público

A propriedade driver.ipTypes contém uma lista separada por vírgulas de tipos de IP na ordem preferida. Por exemplo, driver.ipTypes=PRIVATE,PUBLIC.

Para mais informações, consulte Configurar rede.

Nome da publicação

Por padrão, o conector tenta transmitir de uma publicação chamada dbz_publication. Para especificar outra publicação, adicione publication.name=PUBLICATION_NAME à configuração, em que PUBLICATION_NAME é o nome da publicação. Exemplo: publication.name=my_publication.

Slots de replicação

O PostgreSQL usa slots de replicação para transmitir mudanças na tabela do banco de dados. Por padrão, o conector cria um slot de replicação chamado debezium. Para usar um nome de slot diferente, defina a propriedade slot.name.

Se você criar duas instâncias do conector para o mesmo banco de dados, especifique um nome de slot exclusivo para cada conector.

Por padrão, o conector define a propriedade slot.drop.on.stop como false para evitar a perda de dados. Quando você exclui um conector de forma permanente, é necessário descartar manualmente o slot de replicação que ele estava usando. O nome do slot de replicação é debezium por padrão, a menos que seja configurado de outra forma usando a propriedade slot.name.

Recomendamos que você configure alertas para monitorar o uso do disco WAL no servidor de banco de dados PostgreSQL de origem e descarte todos os slots de replicação não utilizados.

Filtro de tabela

Por padrão, o conector captura dados de mudança de todas as tabelas não sistêmicas no banco de dados. Para filtrar as tabelas capturadas, especifique uma ou mais das seguintes configurações:

  • schema.include.list. Uma lista de esquemas a serem incluídos.
  • schema.exclude.list. Uma lista de esquemas a serem excluídos. Não pode ser usado com schema.include.list.
  • table.include.list. Uma lista de tabelas a serem incluídas.
  • table.exclude.list. Uma lista de tabelas a serem excluídas. Não pode ser usado com table.include.list.

Nomes de tópicos

Por padrão, o conector cria tópicos do Kafka com a seguinte convenção de nomenclatura: topic_prefix.schema.table_name, em que topic.prefix é o valor da configuração topic.prefix.

Para mais informações, consulte Nomes de tópicos na documentação do Debezium.

A seguir