Todos los productos
Search

Este producto

    Realtime Compute for Apache Flink:PostgreSQL CDC

    Centro de documentos

    Realtime Compute for Apache Flink:PostgreSQL CDC

    Última actualización: Sep 25, 2026

    El conector Postgres CDC lee una instantánea completa de una base de datos PostgreSQL y luego captura los datos de cambio, con semántica de procesamiento exactly-once.

    Descripción general

    El conector Postgres CDC es compatible con las siguientes capacidades:

    Categoría

    Detalles

    Tipos compatibles

    Origen SQL, origen Flink CDC

    Nota

    Utilice el conector JDBC para las tablas sink y las tablas de búsqueda (dimensión).

    Modo de ejecución

    Streaming

    Formato de datos

    No aplicable

    Métricas

    Métricas de monitoreo

    • currentFetchEventTimeLag: intervalo desde que se generan los datos hasta que el operador de origen los extrae.

    • currentEmitEventTimeLag: intervalo desde que se generan los datos hasta que salen del operador de origen.

    • sourceIdleTime: duración durante la cual el origen no ha producido datos nuevos.

    Nota

    • Las métricas currentFetchEventTimeLag y currentEmitEventTimeLag solo son válidas en la fase incremental. En la fase de instantánea, su valor siempre es 0.

    • Para obtener más información sobre las métricas, consulte Metric description.

    Tipos de API

    SQL y Flink CDC

    Actualización o eliminación en el destino

    No aplicable

    Funciones

    A partir de VVR 8.0.6, el conector Postgres CDC se integra con el framework de instantáneas incrementales. Lee los datos históricos completos y después cambia automáticamente a la lectura de registros de cambios desde el WAL, con semántica exactly-once.

    Funciones principales:

    • Procesamiento unificado de flujos y lotes. Lee datos completos e incrementales en un solo trabajo.

    • Lecturas de instantáneas concurrentes. Escala horizontalmente para lograr un rendimiento más rápido.

    • Cambio fluido entre la fase completa y la incremental. Reduce automáticamente la escala para disminuir el uso de recursos.

    • Lecturas reanudables. Reanuda la operación desde puntos de interrupción durante la fase de instantánea para mejorar la estabilidad.

    • Lectura sin bloqueos. No requiere bloqueos, lo que evita afectar las operaciones en línea.

    Requisitos previos

    El conector Postgres CDC lee flujos CDC mediante la replicación lógica de PostgreSQL. Es compatible con ApsaraDB RDS for PostgreSQL, Amazon RDS for PostgreSQL y PostgreSQL autogestionado.

    Importante

    La configuración varía según el tipo de despliegue. Configure Postgres.

    Después de configurar el entorno, compruebe lo siguiente:

    • El parámetro wal_level está establecido en logical para habilitar la decodificación lógica.

    • El valor REPLICA IDENTITY de cada tabla suscrita está establecido en FULL, de modo que los eventos INSERT y UPDATE incluyan los valores anteriores de las columnas para garantizar la consistencia de los datos.

      Nota

      REPLICA IDENTITY es una configuración a nivel de tabla en PostgreSQL que controla si los eventos INSERT y UPDATE incluyen los valores anteriores de las columnas. REPLICA IDENTITY.

    • Los valores de max_wal_senders y max_replication_slots superan la suma de las ranuras en uso y las ranuras que requiere el trabajo de Flink.

    • La cuenta tiene privilegios de SUPERUSER, o bien permisos de LOGIN y REPLICATION, además de SELECT sobre las tablas suscritas.

    • Si la tabla de Postgres contiene columnas generadas, establezca el parámetro publish_generated_columns en stored al crear la ranura. De lo contrario, los esquemas de la fase de instantánea y de la fase incremental podrían diferir.

    Notas de uso

    La función de instantánea incremental requiere VVR 8.0.6 o posterior.

    Ranuras de replicación

    Los trabajos de Flink PostgreSQL CDC usan ranuras de replicación para evitar la purga prematura del WAL y garantizar la consistencia de los datos. Una gestión deficiente de estas ranuras puede provocar un uso excesivo del disco o retrasos en la lectura. A continuación se presentan las prácticas recomendadas:

    • Elimine oportunamente las ranuras que no se utilicen

      • Flink no elimina automáticamente las ranuras de replicación cuando un trabajo se detiene o se reinicia sin estado, ya que esto previene la pérdida de datos del WAL.

      • Si un trabajo no se va a reiniciar, elimine manualmente su ranura de replicación para liberar espacio en disco.

        Nota

        Gestión del ciclo de vida: Trate las ranuras de replicación como recursos a nivel de trabajo y gestíonelas junto con los inicios y las detenciones del trabajo.

    • Evite reutilizar ranuras antiguas

      • Utilice siempre un nombre de ranura nuevo. Si reutiliza una ranura antigua, el trabajo leerá los datos históricos acumulados del WAL durante el inicio, lo que retrasará el procesamiento de datos nuevos.

      • PostgreSQL requiere una ranura por conexión. Cada trabajo debe usar un nombre de ranura único.

        Nota

        Convención de nomenclatura: Cuando personalice slot.name, evite nombres con sufijos numéricos como my_slot_1 para prevenir conflictos con ranuras temporales.

    • Comportamiento de las ranuras con instantáneas incrementales habilitadas

      • Requisitos previos: Los puntos de control deben estar habilitados y la tabla de origen debe tener definida una clave principal.

      • Reglas de creación de ranuras:

        • Instantánea incremental deshabilitada: Solo se admite un paralelismo de 1. Se utiliza una ranura global.

        • Instantánea incremental habilitada:

          • Fase de instantánea: Cada subtarea de origen concurrente crea una ranura temporal. El formato de nomenclatura es ${slot.name}_${task_id}.

          • Fase incremental: Todas las ranuras temporales se recuperan automáticamente. Solo se conserva una ranura global.

      • Número máximo de ranuras: Paralelismo del origen + 1 (durante la fase de instantánea).

    • Recursos y rendimiento

      • Si las ranuras disponibles o el espacio en disco son limitados, reduzca el paralelismo de la instantánea para usar menos ranuras temporales. Esto disminuye la velocidad de lectura de la instantánea.

      • Si el destino descendente admite escrituras idempotentes, establezca scan.incremental.snapshot.backfill.skip = true para omitir el llenado retroactivo del WAL durante la fase de instantánea y acelerar el inicio.

        Esta opción solo proporciona semántica at-least-once y no es adecuada para cálculos con estado (agregaciones o uniones de búsqueda), porque pueden perderse los cambios históricos necesarios.

    • Cuando las instantáneas incrementales están deshabilitadas, no se admiten puntos de control durante la fase de instantánea.

      Configuración para evitar tiempos de espera agotados durante la fase de instantánea

      Cuando las instantáneas incrementales están deshabilitadas, los puntos de control durante la fase de instantánea pueden causar failovers por tiempo de espera agotado. Configure estos parámetros en Other Configuration (Configure custom running parameters):

      execution.checkpointing.interval: 10min
execution.checkpointing.tolerable-failed-checkpoints: 100
restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: 2147483647

      Parámetros:

      Parámetro

      Descripción

      Observaciones

      execution.checkpointing.interval

      Intervalo entre puntos de control.

      La unidad es un valor de duración, como 10min o 30s.

      execution.checkpointing.tolerable-failed-checkpoints

      Número de errores de punto de control que se toleran antes de que falle el trabajo.

      El producto de este parámetro y el intervalo de programación del punto de control determina el tiempo permitido para la lectura de la instantánea.

      Nota

      Si la tabla es muy grande, establezca este parámetro en un valor mayor.

      restart-strategy

      Estrategia de reinicio del trabajo.

      Valores válidos:

      • fixed-delay: estrategia de reinicio con retardo fijo.

      • failure-rate: estrategia de reinicio basada en la tasa de errores.

      • exponential-delay: estrategia de reinicio con retardo exponencial.

      Restart Strategies.

      restart-strategy.fixed-delay.attempts

      Número máximo de intentos de reinicio para la estrategia fixed-delay.

      –

    Reutilización de la suscripción de Postgres

    El conector Postgres CDC depende de una publicación para determinar qué cambios de tablas se envían a una ranura. Si varios trabajos comparten la misma publicación, sus configuraciones se sobrescribirán.

    Causa

    El valor predeterminado de publication.autocreate.mode es filtered, que incluye únicamente las tablas definidas en la configuración del conector. Esto modifica la publicación al iniciar el trabajo, lo cual puede afectar a otros trabajos.

    Solución

    1. Cree una publicación en PostgreSQL que incluya todas las tablas monitoreadas, o bien cree una publicación independiente por cada trabajo.

      -- Create a publication named my_flink_pub that includes all tables (or specified tables, creating one publication per job)
CREATE PUBLICATION my_flink_pub FOR TABLE table_a, table_b;
-- Or more simply, include all tables in the database
CREATE PUBLICATION my_flink_pub FOR ALL TABLES;
      Nota

      No se recomienda suscribirse a todas las tablas en bases de datos grandes, debido al consumo excesivo de ancho de banda y CPU en el clúster de Flink.

    2. Agregue las siguientes configuraciones de Flink:

      • debezium.publication.name = 'my_flink_pub' (Especifica el nombre de la publicación)

      • debezium.publication.autocreate.mode = 'disabled' (Impide que Flink intente crear o modificar la publicación durante el inicio)

    Esto garantiza un aislamiento completo y evita que los trabajos nuevos afecten a los existentes.

    SQL

    Sintaxis

    CREATE TABLE postgrescdc_source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>',
  'decoding.plugin.name'= 'pgoutput',
  'scan.incremental.snapshot.enabled' = 'true',
  -- Skipping backfill can speed up reads and reduce resource usage, but it may cause data duplication. Enable this if the downstream sink is idempotent.
  'scan.incremental.snapshot.backfill.skip' = 'false',
  -- In a production environment, set this to 'filtered' or 'disabled' and manage the publication manually instead of through Flink.
  'debezium-publication.autocreate.mode' = 'disabled'
  -- If you have multiple sources, configure a different publication for each source.
  --'debezium.publication.name' = 'my_flink_pub'
);

    Opciones del conector

    Opción

    Descripción

    Tipo de datos

    Obligatorio

    Predeterminado

    Observaciones

    connector

    Nombre del conector.

    STRING

    Sí

    –

    El valor debe ser postgres-cdc.

    hostname

    Dirección IP o nombre de host de la base de datos PostgreSQL.

    STRING

    Sí

    –

    –

    username

    Nombre de usuario del servicio de base de datos PostgreSQL.

    STRING

    Sí

    –

    –

    password

    Contraseña del servicio de base de datos PostgreSQL.

    STRING

    Sí

    –

    –

    database-name

    Nombre de la base de datos PostgreSQL.

    STRING

    Sí

    –

    Nombre de la base de datos.

    schema-name

    Nombre del esquema de PostgreSQL. Admite expresiones regulares.

    STRING

    Sí

    –

    El nombre del esquema admite expresiones regulares para leer datos de varios esquemas.

    table-name

    Nombre de la tabla de PostgreSQL. Admite expresiones regulares.

    STRING

    Sí

    –

    El nombre de la tabla admite expresiones regulares para leer datos de varias tablas.

    port

    Número de puerto.

    INTEGER

    No

    5432

    –

    decoding.plugin.name

    Nombre del plugin de decodificación lógica de PostgreSQL.

    STRING

    No

    decoderbufs

    Lo determina el plugin instalado en el servicio de PostgreSQL. Los plugins compatibles son:

    • decoderbufs: Compatible con PostgreSQL 9.6 y versiones posteriores. Este plugin debe instalarse.

    • pgoutput (recomendado): Plugin integrado oficial para PostgreSQL 10 y versiones posteriores.

    slot.name

    Nombre de la ranura de decodificación lógica.

    STRING

    Obligatorio para VVR 8.0.1 y versiones posteriores. Opcional para versiones anteriores.

    flink (Anterior a 8.0.1)

    Establezca un valor único de slot.name para cada tabla y así evitar el error PSQLException: ERROR: replication slot "debezium" is active for PID 974. Replication slots.

    No hay valor predeterminado para VVR 8.0.1 y versiones posteriores.

    debezium.*

    Propiedades y parámetros de Debezium

    STRING

    No

    –

    Proporciona un control más detallado sobre el comportamiento del cliente Debezium. Por ejemplo, 'debezium.snapshot.mode' = 'never'. Configuration properties.

    scan.incremental.snapshot.enabled

    Especifica si se habilitan las instantáneas incrementales.

    BOOLEAN

    No

    false
    Nota

    • Esta es una función experimental disponible en VVR 8.0.6 y versiones posteriores.

    • Ventajas, requisitos previos y límites: Features, Prerequisites y Usage notes.

    scan.startup.mode

    Modo de inicio para el consumo de datos.

    STRING

    No

    initial

    Valores válidos:

    • initial: Analiza todos los datos históricos en el primer inicio y luego lee los datos más recientes del WAL.

    • latest-offset: No analiza los datos históricos completos en el primer inicio. Comienza a leer desde el final del WAL, lo que significa que solo lee los cambios más recientes realizados después de que se inicia el conector.

    • snapshot: Analiza los datos históricos completos, lee los datos nuevos del WAL generados durante la fase de instantánea y luego el trabajo se detiene.

    changelog-mode

    Modo de registro de cambios para codificar las modificaciones del flujo.

    String

    No

    all

    Modos de registro de cambios compatibles:

    • ALL: Admite todos los tipos, incluidos INSERT, DELETE, UPDATE_BEFORE y UPDATE_AFTER.

    • UPSERT: Admite únicamente el tipo upsert, que incluye INSERT, DELETE y UPDATE_AFTER.

    heartbeat.interval.ms

    Intervalo para enviar paquetes de latido.

    Duration

    No

    30s

    La unidad es milisegundos.

    El conector Postgres CDC envía activamente latidos a la base de datos para avanzar el desplazamiento de la ranura. Cuando los cambios en la tabla son poco frecuentes, establecer este valor garantiza la recuperación oportuna de los registros del WAL.

    scan.incremental.snapshot.chunk.key-column

    Especifica una columna que se utilizará como clave de fragmento para dividir los shards durante la fase de instantánea.

    STRING

    No

    –

    De forma predeterminada, se selecciona la primera columna de la clave principal.

    scan.incremental.close-idle-reader.enabled

    Especifica si se cierran los lectores inactivos una vez completada la instantánea.

    Boolean

    No

    false

    Para habilitar esta configuración, establezca execution.checkpointing.checkpoints-after-tasks-finish.enabled en true.

    scan.incremental.snapshot.backfill.skip

    Especifica si se omite la lectura de registros durante la fase de instantánea.

    Boolean

    No

    false

    Valores válidos:

    • true: omite la lectura.

      En la fase incremental, los registros se leen a partir de la marca de agua baja.

      Si los operadores descendentes o el almacenamiento admiten idempotencia, se recomienda omitir la lectura de registros en la fase completa. Esto reduce la cantidad de ranuras del WAL, aunque solo se puede garantizar la semántica at-least-once.

    • false: no omite la lectura.

      Al leer las divisiones en la fase completa, se leen los registros comprendidos entre la marca de agua baja y la alta para garantizar la consistencia.

      Si la instrucción SQL ejecuta agregaciones, uniones u operaciones similares, no se recomienda omitir la lectura de registros en la fase completa.

    Mapeo de tipos

    A continuación se muestra el mapeo de tipos de PostgreSQL a Flink:

    PostgreSQL CDC

    Flink

    SMALLINT

    SMALLINT

    INT2

    SMALLSERIAL

    SERIAL2

    INTEGER

    INT

    SERIAL

    BIGINT

    BIGINT

    BIGSERIAL

    REAL

    FLOAT

    FLOAT4

    FLOAT8

    DOUBLE

    DOUBLE PRECISION

    NUMERIC(p, s)

    DECIMAL(p, s)

    DECIMAL(p, s)

    BOOLEAN

    BOOLEAN

    DATE

    DATE

    TIME [(p)] [WITHOUT TIMEZONE]

    TIME [(p)] [WITHOUT TIMEZONE]

    TIMESTAMP [(p)] [WITHOUT TIMEZONE]

    TIMESTAMP [(p)] [WITHOUT TIMEZONE]

    CHAR(n)

    STRING

    CHARACTER(n)

    VARCHAR(n)

    CHARACTER VARYING(n)

    TEXT

    BYTEA

    Bytes

    Ejemplo

    CREATE TABLE source (
  id INT NOT NULL,
  name STRING,
  description STRING,
  weight DECIMAL(10,3)
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = '<host name>',
  'port' = '<port>',
  'username' = '<user name>',
  'password' = '<password>',
  'database-name' = '<database name>',
  'schema-name' = '<schema name>',
  'table-name' = '<table name>'
);

SELECT * FROM source;

    Flink CDC

    VVR V11.4 y versiones posteriores son compatibles con el conector de PostgreSQL como origen de Flink CDC.

    Sintaxis

    source:
  type: postgres
  name: PostgreSQL Source
  hostname: localhost
  port: 5432
  username: pg_username
  password: pg_password
  tables: db.scm.tbl
  slot.name: test_slot
  scan.startup.mode: initial
  server-time-zone: UTC
  connect.timeout: 120s
  decoding.plugin.name: decoderbufs

sink:
  type: ...

    Opciones del conector

    Opción

    Descripción

    Obligatorio

    Tipo de datos

    Predeterminado

    Observaciones

    type

    Nombre del conector.

    Sí

    STRING

    –

    Debe ser postgres.

    name

    Nombre del origen de datos.

    No

    STRING

    –

    –

    hostname

    Nombre de dominio o dirección IP del servidor de la base de datos PostgreSQL.

    Sí

    STRING

    –

    –

    port

    Puerto de la base de datos PostgreSQL.

    No

    INTEGER

    5432

    –

    username

    Nombre de usuario de PostgreSQL.

    Sí

    STRING

    –

    –

    password

    Contraseña de PostgreSQL.

    Sí

    STRING

    –

    –

    tables

    Nombres de las tablas que se capturarán.

    Compatible con expresiones regulares.

    Sí

    STRING

    –
    Importante

    Actualmente, solo se pueden capturar tablas de la misma base de datos.

    El punto (.) se interpreta como separador de un nombre completo. Para usarlo en una expresión regular y coincidir con cualquier carácter, escápelo con una barra invertida. Por ejemplo: bdb.schema_\..order_\..

    slot.name

    Nombre del slot de replicación de PostgreSQL.

    Sí

    STRING

    –

    El nombre debe cumplir con las reglas de nomenclatura de slots de replicación de PostgreSQL. Puede contener letras minúsculas, números y guiones bajos.

    decoding.plugin.name

    Nombre del plugin de decodificación lógica de PostgreSQL instalado en el servidor.

    No

    STRING

    pgoutput

    Valores válidos: decoderbufs y pgoutput.

    tables.exclude

    Tablas que se excluirán. Esta opción surte efecto después de la opción tables. Compatible con expresiones regulares.

    No

    STRING

    –

    Consulte la opción tables.

    server-time-zone

    Zona horaria de la sesión del servidor de la base de datos, por ejemplo, "Asia/Shanghai".

    No

    STRING

    –

    Si no se establece, se utiliza la zona horaria predeterminada del sistema (ZoneId.systemDefault()).

    scan.incremental.snapshot.chunk.size

    Tamaño (número de filas) de cada fragmento en el framework de instantáneas incrementales.

    No

    INTEGER

    8096

    Cuando se habilita la instantánea incremental, la tabla se divide en varios fragmentos para su lectura. Los datos de un fragmento se almacenan en caché en la memoria hasta que se consumen por completo.

    Un fragmento más pequeño aumenta el número total de fragmentos de la tabla. Esto reduce la granularidad de la recuperación ante fallos, pero puede provocar errores de falta de memoria (OOM) y disminuir el throughput general. Por lo tanto, busque un equilibrio y establezca un tamaño de fragmento adecuado.

    scan.snapshot.fetch.size

    Número máximo de registros que se obtienen a la vez al leer los datos completos de una tabla.

    No

    INTEGER

    1024

    –

    scan.startup.mode

    Modo de inicio para el consumo de datos.

    No

    STRING

    initial

    Valores válidos:

    • initial (predeterminado): analiza la instantánea en el primer inicio y luego cambia a los datos WAL más recientes.

    • latest-offset: omite la lectura de la instantánea e inicia la lectura desde el final del WAL. Es decir, solo lee los cambios más recientes realizados después de que se inicia el conector.

    • committed-offset: omite la lectura de la instantánea y consume los datos WAL desde un desplazamiento específico.

    • snapshot: consume únicamente la instantánea, sin procesar los datos incrementales.

    scan.incremental.close-idle-reader.enabled

    Indica si se cierran los lectores inactivos una vez que finaliza la instantánea.

    No

    BOOLEAN

    false

    Para habilitar esta configuración, establezca execution.checkpointing.checkpoints-after-tasks-finish.enabled en true.

    scan.lsn-commit.checkpoints-num-delay

    Número de puntos de control que se retrasan antes de comenzar a confirmar los desplazamientos LSN.

    No

    INTEGER

    3

    Los desplazamientos LSN de los puntos de control se confirman de forma rotativa para evitar que sea imposible recuperar el estado.

    connect.timeout

    Tiempo máximo que el conector espera para conectarse al servidor de la base de datos PostgreSQL antes de agotar el tiempo de espera.

    No

    DURATION

    30s

    Este valor no puede ser inferior a 250 milisegundos.

    connect.max-retries

    Número máximo de intentos de reconexión del conector para establecer una conexión.

    No

    INTEGER

    3

    –

    connection.pool.size

    Tamaño del pool de conexiones.

    No

    INTEGER

    20

    –

    jdbc.properties.*

    Permite pasar propiedades personalizadas de la URL JDBC.

    No

    STRING

    20

    Se pueden pasar propiedades personalizadas, por ejemplo, 'jdbc.properties.useSSL' = 'false'.

    heartbeat.interval

    Intervalo para enviar eventos de latido que rastrean el último desplazamiento disponible del registro WAL.

    No

    DURATION

    30s

    –

    debezium.*

    Pasa las propiedades de Debezium al motor integrado de Debezium, que se utiliza para capturar los cambios de datos del servidor PostgreSQL.

    No

    STRING

    –

    Propiedades del conector Debezium para PostgreSQL: documentación de Debezium.

    chunk-meta.group.size

    Tamaño de los metadatos del fragmento.

    No

    STRING

    1000

    Si los metadatos superan este valor, se transmiten en varias partes.

    metadata.list

    Lista de metadatos legibles que se pasan al flujo descendente. Se pueden utilizar en el módulo de transformación.

    No

    STRING

    false

    Utilice comas (,) como separadores. A continuación se muestran los metadatos disponibles:

    • op_ts

    • table_name

    • database_name

    • schema_name

    Nota

    Las columnas de metadatos están disponibles únicamente en VVR 11.9 y versiones posteriores.

    scan.incremental.snapshot.unbounded-chunk-first.enabled

    Envía primero el fragmento ilimitado durante la fase de lectura de la instantánea.

    No

    STRING

    false

    Esta es una función experimental. Habilitarla reduce el riesgo de errores OOM cuando TaskManager sincroniza el último fragmento durante la fase de instantánea. Se recomienda agregarla antes del primer inicio del trabajo.

    schema-change.enabled

    Indica si el conector deduce automáticamente los cambios de esquema.

    No

    BOOLEAN

    false

    Una vez habilitada esta función, el conector compara los mensajes recibidos por pgoutput con el esquema de la tabla almacenado en caché para deducir los eventos de cambio de esquema.

    Nota

    Los cambios de esquema que se pueden deducir incluyen agregar columnas, eliminar columnas, cambiar el nombre de columnas y modificar los tipos de datos de las columnas.

    Este parámetro está disponible únicamente en VVR 11.9 y versiones posteriores.

    Referencias