Crea un conector de origen de Cloud SQL para PostgreSQL

En este documento, se describe cómo crear un conector de origen de Cloud SQL para PostgreSQL para Kafka Connect.

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

Estos son algunos casos de uso de este conector:

  • Supervisa los cambios en la base de datos a nivel de la fila en tiempo real.
  • Integra eventos de cambio de la base de datos en una arquitectura basada en eventos.
  • Responder a eventos de bases de datos, como inserciones o eliminaciones de filas
  • Copiar los cambios de la base de datos a otros sistemas

Antes de comenzar

Antes de crear un conector de origen de Cloud SQL para PostgreSQL, asegúrate de tener lo siguiente:

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 administrados (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:

  • Crea un conector: managedkafka.connectors.create

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

Cómo otorgar permisos para leer desde Cloud SQL

La cuenta de servicio de Kafka administrado debe tener permiso para acceder a Cloud SQL para PostgreSQL. Otorga los siguientes roles de IAM a la cuenta de servicio:

  • Cliente de Cloud SQL (roles/cloudsql.client)
  • Usuario de instancia de Cloud SQL (roles/cloudsql.instanceUser)

La cuenta de servicio de Kafka administrado tiene el siguiente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka., en el que PROJECT_NUMBER es el número del proyecto del clúster de Connect.

Si tu clúster de Connect se encuentra en un proyecto diferente del clúster de Managed Service para Apache Kafka, consulta cómo crear un clúster de Connect en un proyecto diferente.

Configura la base de datos

Antes de crear el conector, debes configurar la replicación de la base de datos y habilitar el conector para que se autentique con la base de datos. En las siguientes secciones, se describen estos pasos.

Habilita la decodificación lógica

Un conector de fuente de Cloud SQL para PostgreSQL depende de la función de decodificación lógica de PostgreSQL. Para habilitar la decodificación lógica en tu instancia de Cloud SQL para PostgreSQL, sigue estos pasos.

Console

  1. Ve a Cloud SQL > Instancias.

    Ir a Instancias

  2. Haz clic en el nombre de la instancia .

  3. Haz clic en Editar.

  4. Expande Parámetros y marcas.

  5. Haz clic en Agregar una marca de base de datos.

  6. En la lista Choose a flag, selecciona cloudsql.logical_decoding.

  7. En Valor, selecciona On.

  8. Haz clic en Listo.

  9. Haz clic en Guardar.

Para obtener más información, consulta Configura la replicación y decodificación lógicas.

Configura la captura de datos modificados (CDC)

Después de habilitar la decodificación lógica en tu instancia, habilita la captura de datos modificados (CDC) para las tablas que quieras replicar.

Para habilitar la CDC para una tabla, ejecuta la instrucción de SQL CREATE PUBLICATION. Esta instrucción crea una publicación, que define un grupo de tablas para replicar.

  • Opción 1. Crea una publicación que replique los cambios de todas las tablas de la base de datos.

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

  • Opción 2 Crea una publicación para un conjunto específico de tablas.

    CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;

    Reemplaza TABLE_LIST por una lista separada por comas de tablas, en el formato "schema_name"."table_name". Incluir los nombres del esquema y de la tabla entre comillas dobles, como se muestra, evita errores de sintaxis si los nombres contienen caracteres especiales o letras mayúsculas.

De forma predeterminada, el conector usa dbz_publication para el nombre de publicación. Para usar una publicación con un nombre diferente, consulta Nombre de la publicación.

Crea una cuenta de usuario para la cuenta de servicio de Kafka administrado

El conector de origen de Cloud SQL para PostgreSQL usa la autenticación de IAM para bases de datos para conectarse a la base de datos. Para habilitar la autenticación de la base de datos de IAM, agrega la cuenta de servicio de Managed Kafka a la instancia de Cloud SQL de la siguiente manera:

Console

  1. Ve a Cloud SQL > Instancias.

    Ir a Instancias

  2. Haz clic en el nombre de la instancia .

  3. En el panel de navegación, haz clic en Usuarios.

  4. Haz clic en Agregar cuenta de usuario.

  5. En el panel Agregar una cuenta de usuario, selecciona Cloud IAM.

  6. En el campo Principal de IAM, ingresa lo siguiente:

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    Reemplaza PROJECT_NUMBER por el número de proyecto del clúster de Connect.

  7. Haz clic en Agregar.

gcloud

Ejecuta el comando gcloud sql users create:

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

Reemplaza lo siguiente:

  • PROJECT_NUMBER: Es el número del proyecto del clúster de Connect.

  • INSTANCE_NAME: Es el nombre de la instancia de Cloud SQL para PostgreSQL.

Debido al límite de longitud en el nombre de usuario de una base de datos, se quita el sufijo . del nombre de usuario, por lo que el nombre de usuario es service-PROJECT_NUMBER@gcp-sa-managedkafka.iam. Cuando ejecutes consultas SQL que hagan referencia a la cuenta de usuario de IAM, especifica el nombre truncado.

Configura la cuenta de usuario

Después de crear la cuenta de usuario de IAM, conéctate a la base de datos como un usuario con el rol cloudsqlsuperuser (como el usuario postgres predeterminado) y ejecuta las siguientes consultas de SQL.

Console

  1. Permite al usuario leer el registro de escritura anticipada.

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

  2. Otorga al usuario permiso SELECT en las tablas.

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

    Como alternativa, puedes otorgar el permiso SELECT en tablas individuales. Si eliges esta opción, también debes establecer la propiedad de configuración table.include.list del conector en la lista de tablas permitidas. La siguiente consulta en SQL otorga permiso de SELECT en una sola tabla:

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

  3. Para cada tabla, otorga al usuario acceso al esquema de la tabla. Puedes omitir este paso si la tabla está en el esquema public predeterminado.

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

Configura las redes

Un conector de origen de Cloud SQL para PostgreSQL puede conectarse a la instancia de Cloud SQL de las siguientes maneras:

  • IP privada
  • Private Service Connect
  • IP pública

Para obtener más información sobre estas opciones, consulta Elige cómo conectarte a Cloud SQL. Como práctica recomendada de seguridad, se recomienda usar una IP privada o Private Service Connect, ya que estas opciones no requieren conectarse a una dirección IP externa.

En la siguiente tabla, se muestran los requisitos de red para cada opción:

Tipo de dirección IP Requisitos
IP privada Configura una IP privada para tu instancia. Para obtener más información, consulta Configura IP privadas.
Private Service Connect
  1. Configura Private Service Connect para tu instancia y obtén el nombre de DNS del extremo de Private Service Connect. Para obtener más información, consulta Cómo conectarse a una instancia con Private Service Connect.
  2. Agrega el nombre de DNS del extremo a los dominios de DNS que se pueden resolver del clúster de Connect. Para obtener más información, consulta Actualiza un clúster de Connect.
IP pública
  1. Configura una IP pública para tu instancia. Para obtener más información, consulta Configura la IP pública.
  2. Configura la NAT pública para permitir que los trabajadores del clúster de Connect se comuniquen con Internet. Para obtener más información, consulta Configura NAT pública. Cuando crees la puerta de enlace de Cloud NAT, especifica la red de VPC que contiene la subred principal del clúster de Connect.

Crea un conector de origen de Cloud SQL para PostgreSQL

Para crear un conector de fuente de Cloud SQL para PostgreSQL, 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 el conector se ejecuta, sigue 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 consola de Google Cloud , ve a la página Connect Clusters.

    Ir a Connect Clusters

  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 Complemento del conector, selecciona Fuente de Cloud SQL para PostgreSQL.

  6. En la lista Instancia, selecciona la instancia de Cloud SQL.

  7. En la lista Base de datos, selecciona la base de datos de Cloud SQL.

  8. En el campo Prefijo del tema, ingresa un prefijo para usar en los nombres de los temas de Kafka. Elige un prefijo único para cada conector de origen de Cloud SQL para PostgreSQL.

  9. Opcional: En el campo Nombres de tablas, ingresa una lista separada por comas de las tablas desde las que se leerán los datos de cambio, en el formato "schema_name"."table_name". Si dejas este campo vacío, el conector leerá los datos de cambio de todas las tablas que no sean del sistema en la base de datos.

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

    Es posible que debas anular los valores predeterminados de las siguientes propiedades:

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

  12. Haz clic en Crear.

gcloud

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

    Activa Cloud Shell

    En la parte inferior de la consola de Google 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 comando gcloud managed-kafka connectors create:

    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 origen de 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"

    Reemplaza lo siguiente:

    • INSTANCE_ID: Es el ID de la instancia de Cloud SQL que contiene la base de datos, con el siguiente formato:

      PROJECT_ID:REGION:INSTANCE_NAME

    • DATABASE_NAME: Es el nombre de la base de datos de Cloud SQL desde la que se leerá.

    • IP_TYPES: Es una lista separada por comas de tipos de direcciones IP.

    • SLOT_NAME: Es el nombre de la ranura de replicación que se creará.

    • TABLE_LIST: Es una lista separada por comas de las tablas desde las que se leerán los datos de cambio, en el formato "schema_name"."table_name".

    • TOPIC_PREFIX: Es un prefijo que se usará para 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 el conector de Debezium para PostgreSQL en la documentación de Debezium.

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 orden de preferencia; por ejemplo, driver.ipTypes=PRIVATE,PUBLIC.

Para obtener más información, consulta Configura la red.

Nombre de la publicación

De forma predeterminada, el conector intenta transmitir desde una publicación llamada dbz_publication. Para especificar otra publicación, 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 en las tablas 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, configura 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 propiedad slot.drop.on.stop en false para evitar la pérdida de datos. Cuando borras un conector de forma permanente, debes descartar manualmente la ranura de replicación que usaba el conector. El nombre de la ranura de replicación se establece de forma predeterminada en debezium, a menos que se configure de otra manera con la propiedad slot.name.

Te recomendamos que configures alertas para supervisar el uso del disco de WAL en el servidor de la base de datos PostgreSQL de origen y que descartes las ranuras de replicación que no se usen.

Filtro de tabla

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

  • schema.include.list: Es una lista de esquemas que se incluirán.
  • schema.exclude.list: Es una lista de esquemas que se excluirán. No se puede usar con schema.include.list.
  • table.include.list: Es una lista de tablas que se incluirán.
  • table.exclude.list: Es 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?