Criar um conector de origem genérico do PostgreSQL

Um conector de origem genérica do PostgreSQL permite capturar alterações no nível da linha de um banco de dados PostgreSQL e transmiti-las para tópicos em um cluster do Serviço Gerenciado para Apache Kafka.

Os casos de uso comuns desse conector incluem:

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

Para ler as mudanças de um banco de dados do Cloud SQL para PostgreSQL, use o conector de origem do Cloud SQL para PostgreSQL.

Antes de começar

Antes de criar um conector de origem genérico do PostgreSQL, verifique se você tem o seguinte:

  • Um banco de dados PostgreSQL.

  • Um cluster do Connect associado ao cluster do Kafka.

  • Crie um secret do Secret Manager que armazene a senha do banco de dados. Se a configuração usar SSL de banco de dados, crie também um secret para a senha de SSL do banco de dados. Configure o cluster do Connect com os secrets. Para mais informações, consulte Recursos do Secret Manager.

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 de conector do Kafka gerenciado (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 acessar secrets do Secret Manager

A conta de serviço do Managed Kafka precisa de permissão para visualizar e acessar os secrets armazenados no Secret Manager. Conceda os seguintes papéis do IAM à conta de serviço:

  • Leitor do Secret Manager (roles/secretmanager.viewer)
  • Acessador de secrets do Secret Manager (roles/secretmanager.secretAccessor)

A conta de serviço do Managed Kafka tem o seguinte formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, 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.

Configure o banco de dados do PostgreSQL

Para permitir que o conector leia eventos de mudança de dados do seu banco de dados, defina as seguintes configurações.

  1. Defina o wal_level do servidor como logical.

    ALTER SYSTEM SET wal_level = logical;

    Reinicie o servidor para aplicar a configuração.

  2. Crie um usuário de banco de dados para o conector se autenticar no PostgreSQL. O usuário do banco de dados precisa ter um papel de replicação, permitindo que ele se conecte ao servidor no modo de replicação.

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    Substitua:

    • ROLE_NAME: o nome do usuário, por exemplo, debezium_user.
    • ROLE_PASSWORD: a senha do usuário.

  3. Crie uma publicação para as tabelas que você quer capturar. O conector se inscreve na publicação para receber eventos de mudança de dados.

    CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";

    Substitua:

    • SCHEMA_NAME: o esquema da tabela.

    • TABLE_NAME: o nome da tabela.

    Recomendamos colocar os nomes do esquema e da tabela entre aspas duplas, como mostrado, para evitar erros de sintaxe se os nomes tiverem caracteres especiais ou letras maiúsculas.

    Como alternativa, é possível criar uma publicação que replique as alterações de todas as tabelas no banco de dados:

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    Dependendo da configuração publication.autocreate.mode do conector, é possível criar a publicação manualmente ou deixar que o conector a crie automaticamente. Para mais informações, consulte Modo de publicação.

  4. Para cada tabela, conceda privilégios SELECT na tabela ao usuário do banco de dados.

    GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;

    Como alternativa, é possível conceder a seleção em todas as tabelas de um esquema:

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

  5. Para cada tabela, conceda privilégios USAGE no esquema da tabela ao usuário do banco de dados. Pule esta etapa se a tabela estiver no esquema padrão public.

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

Criar um conector de origem genérico do PostgreSQL

Para criar um conector de origem genérico do PostgreSQL, siga as etapas abaixo.

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 alteração para o tópico correspondente.

Enquanto o conector estiver em execução, ele vai continuar enviando eventos de alteração 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 Clusters do Connect.

    Acessar clusters do Connect

  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 genérica do PostgreSQL.

  6. No campo Nome do host do banco de dados, insira o nome do host ou o endereço IP do servidor PostgreSQL.

  7. No campo Nome do banco de dados, insira o nome do banco de dados.

  8. No campo Usuário do banco de dados, insira o nome do papel de réplica. O conector se autentica no servidor PostgreSQL usando esse papel.

  9. No campo Prefixo do tópico, insira um prefixo para usar nos nomes dos tópicos do Kafka.

  10. Na lista Secret, selecione o secret que contém a senha do banco de dados.

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

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

  13. Clique em Criar.

gcloud

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

    Ativar o Cloud Shell

    Na parte inferior do console do 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. Esse local precisa ser o mesmo 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 PostgreSQL genérico:

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
database.hostname: HOSTNAME
database.password: CREDENTIALS
database.user: DATABASE_USER
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    Substitua:

    • HOSTNAME: o nome do host do banco de dados do PostgreSQL de onde os dados serão lidos.

    • DATABASE_NAME: o nome do banco de dados do PostgreSQL de onde os dados serão lidos.

    • DATABASE_USER: o usuário do banco de dados PostgreSQL a ser usado na autenticação no banco de dados.

    • CREDENTIALS: um caminho para o secret do Secret Manager que contém a senha do banco de dados. Especifique o secret usando o seguinte formato:

      ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}

    • TOPIC_PREFIX: um prefixo a ser usado para os nomes dos 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 Debezium para PostgreSQL na documentação do Debezium.

Configurações de senha e senha SSL

Somente caminhos de secrets são compatíveis com as configurações database.password e database.sslpassword. O back-end espera que essas configurações usem o seguinte formato: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}.

Tipos de endereço IP

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

  • PRIVATE: IP particular
  • 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 preferencial, por exemplo, driver.ipTypes=PRIVATE,PUBLIC.

Modo de publicação

Um conector de origem genérica do PostgreSQL transmite eventos de alteração de uma publicação no banco de dados. É possível criar a publicação manualmente ou deixar que o conector a crie automaticamente.

A configuração publication.autocreate.mode especifica como e se o conector deve criar uma publicação.

  • filtered. Se a publicação não existir, o conector vai criar uma nova publicação que inclui apenas as tabelas capturadas. O usuário do banco de dados precisa ter permissões CREATE no banco de dados e ser o proprietário das tabelas incluídas.

    Se a publicação já existir, o conector vai alterá-la para incluir as tabelas capturadas. Para alterar uma publicação, o usuário do banco de dados precisa ser o proprietário da publicação e das tabelas incluídas.

  • all_tables. Se a publicação não existir, o conector vai criar uma nova usando o parâmetro FOR ALL TABLES. O usuário do banco de dados precisa ser um superusuário.

    Os papéis de superusuário ignoram todas as verificações de permissão em um banco de dados. Portanto, não é recomendável conceder SUPERUSER ao usuário do banco de dados. Em vez disso, crie a publicação manualmente ou defina publication.autocreate.mode=filtered.

  • disabled. Se a publicação não existir, um erro vai ocorrer. O conector não cria uma nova publicação.

O valor padrão é all_tables.

Nome da publicação

Por padrão, o conector tenta fazer streaming de uma publicação chamada dbz_publication. Para especificar uma publicação diferente, 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. Ao excluir permanentemente um conector, é necessário descartar manualmente o slot de replicação que o conector 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 elimine os slots de replicação não utilizados.

Filtro da tabela

Por padrão, o conector captura dados de alteração de todas as tabelas que não são do sistema no banco de dados. Para filtrar quais tabelas são 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 dos 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

Apache Kafka® é uma marca registrada da The Apache Software Foundation ou afiliadas nos Estados Unidos e/ou em outros países.