Crea un conector de fuente de PostgreSQL genérico

En este documento, se describe cómo crear un conector de fuente de PostgreSQL genérico.

Un conector de fuente de PostgreSQL genérico es una instancia de un conector de PostgreSQL de Debezium. Lee los cambios a nivel de fila de una base de datos de PostgreSQL y los escribe en temas de un clúster de Managed Service para Apache Kafka.

Los casos de uso de este conector incluyen los siguientes:

  • Supervisar los cambios de la base de datos a nivel de fila en tiempo real
  • Integrar eventos de cambio de base de datos en una arquitectura basada en eventos
  • Responder a eventos de bases de datos, como inserciones o eliminaciones de filas
  • Copiar cambios de bases de datos a otros sistemas
  • Replicar o restablecer tablas de PostgreSQL

Antes de comenzar

Antes de crear un conector de fuente de PostgreSQL genérico, asegúrate de tener lo siguiente:

  • Una base de datos de PostgreSQL

  • Un clúster de Connect asociado con el clúster de Kafka.

  • Crea un secreto de Secret Manager que almacene la contraseña de la base de datos. Si tu configuración usa SSL de base de datos, también crea un secreto para la contraseña de SSL de la base de datos. Configura tu clúster de Connect con los secretos. Para obtener más información, consulta Recursos de Secret Manager.

Roles y permisos requeridos

Para obtener los permisos que necesitas para crear un conector, pídele a tu administrador que te otorgue el rol de IAM Editor de conectores de Kafka administrado (roles/managedkafka.connectorEditor) en tu proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear un conector. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear un conector:

  • Crear un conector: managedkafka.connectors.create

También puedes obtener estos permisos con roles personalizados o otros roles predefinidos.

Otorga permisos para acceder a los secretos de Secret Manager

La cuenta de servicio de Kafka administrada necesita permiso para ver y acceder a los secretos almacenados en Secret Manager. Otorga las siguientes funciones de IAM a la cuenta de servicio:

  • Visualizador de Secret Manager (roles/secretmanager.viewer)
  • Descriptor de acceso de secretos de Secret Manager (roles/secretmanager.secretAccessor)

La cuenta de servicio de Kafka administrada tiene el siguiente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, donde PROJECT_NUMBER es el número de proyecto del clúster de Connect.

Si tu clúster de Connect está en un proyecto diferente del clúster de Managed Service para Apache Kafka, consulta Crea un clúster de Connect en un proyecto diferente.

Configura tu base de datos de PostgreSQL

Para permitir que el conector lea eventos de cambios de datos de tu base de datos, configura los siguientes parámetros.

  1. Establece el wal_level del servidor en logical.

    ALTER SYSTEM SET wal_level = logical;

    Reinicia el servidor para aplicar la configuración.

  2. Crea un usuario de base de datos para que el conector se autentique en PostgreSQL. El usuario de la base de datos debe ser un rol de replicación, lo que le permite conectarse al servidor en modo de replicación.

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    Reemplaza lo siguiente:

    • ROLE_NAME: El nombre del usuario, por ejemplo, debezium_user.
    • ROLE_PASSWORD: La contraseña del usuario.

  3. Crea una publicación para las tablas que deseas capturar. El conector se suscribe a la publicación para recibir eventos de cambios de datos.

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

    Reemplaza lo siguiente:

    • SCHEMA_NAME: El esquema de la tabla.

    • TABLE_NAME: El nombre de la tabla.

    Te recomendamos que incluyas los nombres de esquema y de tabla entre comillas dobles, como se muestra, para evitar errores de sintaxis si los nombres contienen caracteres especiales o letras mayúsculas.

    Como alternativa, puedes crear una publicación que replique los cambios de todas las tablas de la base de datos:

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    Según la configuración publication.autocreate.mode del conector, puedes crear la publicación de forma manual o permitir que el conector la cree automáticamente. Para obtener más información, consulta Modo de publicación.

  4. Para cada tabla, otorga privilegios SELECT en la tabla al usuario de la base de datos.

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

    Como alternativa, puedes seleccionar todas las tablas de un esquema:

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

  5. Para cada tabla, otorga privilegios USAGE en el esquema de la tabla al usuario de la base de datos. Puedes omitir este paso si la tabla está en el esquema public predeterminado.

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

Crea un conector de fuente de PostgreSQL genérico

Para crear un conector de fuente de PostgreSQL genérico, sigue estos pasos.

Cuando se inicializa el conector, realiza las siguientes acciones:

  1. Crea una instantánea inicial de la base de datos.
  2. Crea un tema de Kafka para cada tabla que tenga filas.
  3. Para cada fila de la base de datos, envía un evento de cambio al tema correspondiente.

Mientras se ejecuta el conector, continúa enviando eventos de cambio a los temas. Para obtener más información sobre la instantánea inicial, consulta Instantáneas en la documentación de Debezium.

Console

  1. En la Google Cloud consola de, ve a la página Clústeres de Connect.

    Ir a los clústeres de Connect

  2. Haz clic en el clúster de Connect en el que deseas crear el conector.

  3. Haz clic en Crear conector.

  4. Para el nombre del conector, ingresa una cadena.

    Si necesitas ayuda para asignarle un nombre a un conector, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka.

  5. En Plug-in del conector, selecciona Fuente de PostgreSQL genérica.

  6. En el campo Nombre de host de la base de datos, ingresa el nombre de host o la dirección IP del servidor de PostgreSQL.

  7. En el campo Nombre de la base de datos, ingresa el nombre de la base de datos.

  8. En el campo Usuario de la base de datos, ingresa el nombre del rol de réplica. El conector se autentica en el servidor de PostgreSQL con este rol.

  9. En el campo Prefijo del tema, ingresa un prefijo para usar en los nombres de los temas de Kafka.

  10. En la lista Secreto, selecciona el secreto que contiene la contraseña de la base de datos.

  11. Opcional: En el cuadro Configuraciones, agrega propiedades de configuración o edita las propiedades predeterminadas. Para obtener más información, consulta Configura el conector.

  12. Opcional: Selecciona la Política de reinicio de tareas. Para obtener más información, consulta Política de reinicio de tareas.

  13. Haz clic en Crear.

gcloud

  1. En la Google Cloud consola de, activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la Google Cloud consola de Cloud, se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

  2. Ejecuta el gcloud managed-kafka connectors create comando:

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

    Reemplaza lo siguiente:

    A continuación, se muestra un ejemplo de un archivo de configuración para el conector de fuente de 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"

    Reemplaza lo siguiente:

    • HOSTNAME: El nombre de host de la base de datos de PostgreSQL desde la que se leerá.

    • DATABASE_NAME: El nombre de la base de datos de PostgreSQL desde la que se leerá.

    • DATABASE_USER: El usuario de la base de datos de PostgreSQL que se usará cuando se autentique en la base de datos.

    • CREDENTIALS: Una ruta de acceso al secreto de Secret Manager que contiene la contraseña de la base de datos. Especifica el secreto con el siguiente formato:

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

    • TOPIC_PREFIX: Un prefijo para usar en los nombres de los temas de Kafka.

Configura el conector

En esta sección, se describen algunas propiedades de configuración que puedes establecer en el conector. Para obtener una lista completa, consulta Conector de Debezium para PostgreSQL en la documentación de Debezium.

Configuraciones de contraseña y contraseña SSL

Solo se admiten rutas de acceso secretas en las configuraciones database.password y database.sslpassword. El backend espera que estas configuraciones usen el siguiente formato: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}.

Tipos de dirección IP

La propiedad driver.ipTypes especifica el tipo de dirección IP que usa el conector para conectarse a la base de datos:

  • PRIVATE: IP privada
  • PSC: Private Service Connect
  • PUBLIC: IP pública

La propiedad driver.ipTypes contiene una lista separada por comas de tipos de IP en el orden preferido ; por ejemplo, driver.ipTypes=PRIVATE,PUBLIC.

Modo de publicación

Un conector de fuente de PostgreSQL genérico transmite eventos de cambio desde una publicación en la base de datos. Puedes crear la publicación de forma manual o permitir que el conector la cree automáticamente.

El publication.autocreate.mode parámetro especifica cómo y si el conector debe crear una publicación.

  • filtered. Si la publicación no existe, el conector crea una nueva que incluye solo las tablas capturadas. El usuario de la base de datos debe tener permisos CREATE en la base de datos y ser el propietario de las tablas incluidas.

    Si la publicación ya existe, el conector la modifica para incluir las tablas capturadas. Para modificar una publicación existente, el usuario de la base de datos debe ser el propietario de la publicación y de las tablas incluidas.

  • all_tables. Si la publicación no existe, el conector crea una nueva con el parámetro FOR ALL TABLES. El usuario de la base de datos debe ser un superusuario.

    Los roles de superusuario omiten todas las verificaciones de permisos en una base de datos, por lo que no se recomienda otorgar SUPERUSER al usuario de la base de datos. En su lugar, crea la publicación de forma manual o establece publication.autocreate.mode=filtered.

  • disabled. Si la publicación no existe, se produce un error. El conector no crea una publicación nueva.

El valor predeterminado es all_tables.

Nombre de la publicación

De forma predeterminada, el conector intenta transmitir desde una publicación llamada dbz_publication. Para especificar una publicación diferente, agrega publication.name=PUBLICATION_NAME a la configuración, donde PUBLICATION_NAME es el nombre de la publicación. Ejemplo: publication.name=my_publication.

Ranuras de replicación

PostgreSQL usa ranuras de replicación para transmitir los cambios de la tabla de la base de datos. De forma predeterminada, el conector crea una ranura de replicación llamada debezium. Para usar un nombre de ranura diferente, establece la propiedad slot.name.

Si creas dos instancias del conector para la misma base de datos, debes especificar un nombre de ranura único para cada conector.

De forma predeterminada, el conector establece la slot.drop.on.stop propiedad en false para evitar la pérdida de datos. Cuando borras un conector de forma permanente, debes descartar de forma manual la ranura de replicación que usaba el conector. El nombre de la ranura de replicación es debezium de forma predeterminada, a menos que se configure de forma diferente con la propiedad slot.name.

Te recomendamos que configures alertas para supervisar el uso del disco WAL en tu servidor de base de datos de PostgreSQL de origen y descartes las ranuras de replicación sin usar.

Filtro de tabla

De forma predeterminada, el conector captura datos de cambio de cada tabla que no es del sistema en la base de datos. Para filtrar qué tablas se capturan, especifica uno o más de los siguientes parámetros:

  • schema.include.list. Una lista de esquemas que se incluirán.
  • schema.exclude.list. Una lista de esquemas que se excluirán. No se puede usar con schema.include.list.
  • table.include.list. Una lista de tablas que se incluirán.
  • table.exclude.list. Una lista de tablas que se excluirán. No se puede usar con table.include.list.

Nombres de temas

De forma predeterminada, el conector crea temas de Kafka con la siguiente convención de nombres: topic_prefix.schema.table_name, donde topic.prefix es el valor de la configuración topic.prefix.

Para obtener más información, consulta Nombres de temas en la documentación de Debezium.

¿Qué sigue?