Hologres YAML connector for data ingestion - Realtime Compute for Apache Flink

Este tema describe cómo usar el conector de Hologres para sincronizar datos en un despliegue de ingesta de datos YAML.

Contexto

Hologres es un motor de almacén de datos en tiempo real que admite escrituras, actualizaciones y análisis a gran escala en tiempo real. Hologres es compatible con el protocolo PostgreSQL y admite SQL estándar. Admite el procesamiento analítico en línea (OLAP) y consultas ad hoc sobre petabytes de datos, además de servir datos con alta concurrencia y baja latencia. Hologres se integra con MaxCompute, Realtime Compute for Apache Flink y DataWorks para ofrecer una solución completa de almacén de datos en línea y fuera de línea. A continuación se describen las capacidades del conector YAML de Hologres.

Elemento	Descripción
Tipo de tabla	Sink
Modo de ejecución	Modos streaming y por lotes
Formato de datos	N/A
Métrica	numRecordsOut numRecordsOutPerSecond Nota Para más detalles, consulte Metrics.
Tipo de API	YAML
Actualizaciones o eliminaciones en una tabla sink	Compatible

Funciones

Función	Descripción
Full database synchronization	Sincroniza datos completos e incrementales en tiempo real desde una base de datos completa o varias tablas hacia las tablas sink correspondientes.
Schema change synchronization	Sincroniza los cambios de esquema (como agregar, eliminar o renombrar columnas) desde las tablas de origen hacia las tablas sink correspondientes en tiempo real.
Sharded database and table synchronization	Utilice una expresión regular para coincidir por nombre con las tablas de origen en varias bases de datos fragmentadas. Los datos de estas tablas se combinan y se sincronizan con las tablas sink descendentes que tengan el mismo nombre.
Writing to a partitioned table	Escribe los datos de una tabla ascendente en una tabla particionada de Hologres.
Data type mapping	Asigna los tipos de datos ascendentes a tipos de datos más amplios de Hologres mediante varias estrategias.

Sintaxis

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}

Parámetros

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
type	El tipo de sink.	String	Sí	Ninguna	El valor debe ser `hologres`.
name	El nombre del sink.	String	No	Ninguna	N/A.
dbname	El nombre de la base de datos.	String	Sí	Ninguna	N/A.
username	El nombre de usuario para acceder a la base de datos. Use el AccessKey ID de su cuenta de Alibaba Cloud.	String	Sí	Ninguna	Para más información, consulte How do I view the AccessKey ID and AccessKey secret? Importante Para evitar exponer su AccessKey, especifique sus valores mediante variables. Para más información, consulte Project variables.
password	La contraseña para acceder a la base de datos. Use el AccessKey Secret de su cuenta de Alibaba Cloud.	String	Sí	Ninguna
endpoint	El endpoint del servicio Hologres.	String	Sí	Ninguna	Para más información, consulte Access endpoints.
jdbcRetryCount	El número de reintentos para las operaciones de escritura y consulta cuando falla una conexión.	Integer	No	10	N/A.
jdbcRetrySleepInitMs	El tiempo de espera fijo para cada intento de reintento.	Long	No	1000	Unidad: milisegundos. El tiempo de espera real de un reintento se calcula con la siguiente fórmula: `jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs`.
jdbcRetrySleepStepMs	El tiempo de espera incremental para cada intento de reintento.	Long	No	5000	Unidad: milisegundos. El tiempo de espera real de un reintento se calcula con la siguiente fórmula: `jdbcRetrySleepInitMs + retry * jdbcRetrySleepStepMs`.
jdbcConnectionMaxIdleMs	El tiempo máximo de inactividad de una conexión JDBC.	Long	No	60000	Unidad: milisegundos. Si una conexión permanece inactiva más tiempo que esta duración, se desconecta y se libera.
jdbcMetaCacheTTL	El tiempo de expiración de la información de TableSchema almacenada en la caché local.	Long	No	60000	Unidad: milisegundos.
jdbcMetaAutoRefreshFactor	Un factor que determina cuándo se activa la actualización de la caché. El sistema actualiza automáticamente la caché si su tiempo de vida restante es menor que el tiempo de activación.	Integer	No	4	El tiempo de vida restante de la caché se calcula de la siguiente manera: Tiempo de vida restante de la caché = Tiempo de expiración de la caché - Tiempo que la caché ha estado activa. Después de una actualización automática, el tiempo activo de la caché se restablece a 0. El tiempo de activación se calcula con la fórmula: `jdbcMetaCacheTTL` / `jdbcMetaAutoRefreshFactor`.
mutatetype	El modo de escritura de datos.	String	No	INSERT_OR_UPDATE	Si se establece una clave principal para la tabla física de Hologres, el sink de Hologres garantiza la semántica exactly-once basada en dicha clave. Cuando llegan datos con una clave principal duplicada, el parámetro `mutatetype` determina cómo se actualiza la tabla sink. El parámetro `mutatetype` admite los siguientes valores: INSERT_OR_IGNORE: conserva la primera aparición de los datos e ignora todos los datos posteriores. INSERT_OR_REPLACE: los datos nuevos reemplazan toda la fila existente. INSERT_OR_UPDATE: actualiza un subconjunto de columnas de una fila existente. Por ejemplo, si una tabla tiene cuatro columnas (a, b, c y d) y "a" es la clave principal (PK), al escribir en Hologres solo los datos de las columnas "a" y "b" cuando ya existe una fila con la misma PK, el sistema actualiza únicamente la columna "b". Las columnas "c" y "d" permanecen sin cambios.
createparttable	Indica si se crean automáticamente las particiones inexistentes al escribir en una tabla particionada.	Boolean	No	false	N/A.
sink.delete-strategy	Especifica cómo procesar los mensajes de retracción.	String	No	Ninguna	Valores válidos: IGNORE_DELETE: ignora los mensajes Update Before y Delete. Este valor es adecuado para escenarios donde solo necesita insertar o actualizar datos, pero no eliminarlos. DELETE_ROW_ON_PK: el framework Flink aplica la operación de eliminación según la clave principal. En las operaciones de actualización, primero elimina los datos antiguos y luego inserta los nuevos para garantizar la precisión de los datos.
jdbcWriteBatchSize	En el modo JDBC, el número máximo de registros que se almacenan en búfer en el sink de Hologres antes de una escritura por lotes.	Integer	No	256	Unidad: filas. Nota Los parámetros jdbcWriteBatchSize, jdbcWriteBatchByteSize y jdbcWriteFlushInterval tienen una relación OR. Si configura los tres parámetros, los datos resultantes se escriben en cuanto se cumple cualquiera de las condiciones.
jdbcWriteBatchByteSize	En el modo JDBC, este parámetro especifica el tamaño máximo de datos en bytes que el sink de Hologres almacena en búfer antes de escribir el lote en el destino.	Long	No	2,097,152 bytes (2 MB)	Nota Los parámetros jdbcWriteBatchSize, jdbcWriteBatchByteSize y jdbcWriteFlushInterval tienen una relación OR. Si configura los tres parámetros, los datos resultantes se escriben en cuanto se cumple cualquiera de las condiciones.
jdbcWriteFlushInterval	En el modo JDBC, este parámetro especifica el tiempo máximo que el sink de Hologres espera antes de escribir los datos almacenados en búfer en Hologres.	Long	No	10000	Unidad: milisegundos. Nota Los parámetros jdbcWriteBatchSize, jdbcWriteBatchByteSize y jdbcWriteFlushInterval tienen una relación OR. Si configura los tres parámetros, los datos resultantes se escriben en cuanto se cumple la condición de cualquiera de ellos.
ignoreNullWhenUpdate	Indica si se ignoran los valores nulos de los datos entrantes cuando `mutatetype` se establece en `INSERT_OR_UPDATE`.	Boolean	No	false	Valores válidos: false (predeterminado): escribe los valores nulos en la tabla sink de Hologres. true: ignora los valores nulos de los datos entrantes.
jdbcEnableDefaultForNotNullColumn	Indica si se inserta un valor predeterminado cuando se escribe un valor nulo en una columna NOT NULL que carece de un valor predeterminado definido.	Boolean	No	true	Valores válidos: true (predeterminado): permite que el conector inserte un valor predeterminado. Las reglas son las siguientes: Para una columna de tipo String, se escribe una cadena vacía (""). Para una columna de tipo Number, se escribe 0. Para columnas de tipo Date, timestamp o timestamptz, se escribe 1970-01-01 00:00:00. false: no inserta un valor predeterminado. Se genera una excepción cuando se escribe un valor nulo en una columna NOT NULL.
remove-u0000-in-text.enabled	Indica si se quita el carácter nulo (`\u0000`) de las cadenas antes de escribirlas.	Boolean	No	false	Valores válidos: false (predeterminado): el conector no procesa los datos. Sin embargo, si se encuentran datos incorrectos, una operación de escritura puede generar la siguiente excepción: `ERROR: invalid byte sequence for encoding "UTF8": 0x00` En este caso, debe procesar los datos incorrectos en la tabla de origen o definir la lógica de manejo de datos en su instrucción SQL. true: el conector quita el carácter `\u0000` de las cadenas para evitar excepciones de escritura.
deduplication.enabled	Indica si se realiza la deduplicación dentro de cada lote antes de escribir en los modos jdbc y jdbc_fixed.	Boolean	No	true	Valores válidos: true (predeterminado): habilita la deduplicación dentro de un lote. Si varios registros tienen la misma clave principal, solo se conserva el último. Por ejemplo, considere datos con dos campos, donde el primer campo es la clave principal: Los registros `INSERT (1,'a')` y `INSERT (1,'b')` llegan de forma secuencial. Después de la deduplicación, solo se escribe el registro posterior `(1,'b')` en la tabla sink de Hologres. La tabla sink de Hologres ya contiene el registro `(1,'a')`. Si los registros `DELETE (1,'a')` y `INSERT (1,'b')` llegan de forma secuencial, solo se escribe en Hologres el último registro que llega: `(1,'b')`. Esto se trata como una actualización directa, no como una eliminación seguida de una inserción. false: no se realiza ninguna deduplicación durante la formación de lotes. Si un registro nuevo tiene la misma clave principal que un registro que ya está en el lote actual, primero se escribe el lote actual. Una vez completada la escritura, se procesa el registro nuevo.
sink.type-normalize-strategy	La estrategia de asignación de tipos de datos.	String	No	STANDARD	La estrategia que utiliza el sink de Hologres para convertir los tipos de datos ascendentes a tipos de Hologres. STANDARD: convierte los tipos de Flink CDC a tipos estándar de PostgreSQL (PG). BROADEN: convierte los tipos de Flink CDC a tipos de Hologres de rango más amplio. ONLY_BIGINT_OR_TEXT: convierte todos los tipos de Flink CDC a BIGINT o TEXT en Hologres.
sink.insert.legacy-put-handler	Indica si se usa el Put Handler heredado para escribir datos en Hologres.	Boolean	No	false	Valores válidos: `false` (predeterminado): se usa el nuevo Put Handler para escribir datos. El formato SQL de la operación de escritura es `insert into xxx(c0,c1,...) select unnest(?),unnest(?),... on conflict`. true: escribe datos mediante el Put Handler heredado. El formato SQL de escritura es `insert into xxx(c0,c1,...) values (?,?,...),... on conflict;` .
table_property.*	Las propiedades de la tabla física para Hologres.	String	No	Ninguna	Cuando crea una tabla de Hologres, puede establecer las propiedades de la tabla física en la cláusula WITH. Las propiedades de tabla adecuadas ayudan al sistema a organizar y consultar datos de manera eficiente. Aviso El parámetro `table_property.distribution_key` toma como valor predeterminado el valor de la clave principal. No cambie esta configuración a menos que comprenda completamente el impacto, ya que puede afectar la correcta escritura de los datos.
connection.ssl.mode	Indica si se habilita el cifrado de transporte Secure Sockets Layer (SSL) y qué modo se utiliza.	String	No	disable	`disable` (predeterminado): deshabilita el cifrado de transporte. `require`: habilita SSL para cifrar el enlace de datos. `verify-ca`: habilita SSL, cifra el enlace de datos y usa un certificado de CA para verificar la autenticidad del servidor de Hologres. `verify-full`: habilita SSL, cifra el enlace de datos, usa un certificado de CA para verificar la autenticidad del servidor de Hologres y comprueba que el nombre CN o DNS del certificado coincida con el endpoint de Hologres configurado. Nota Hologres V2.1 y versiones posteriores admiten los modos verify-ca y verify-full. Para más información, consulte Transport encryption. Si establece este parámetro en `verify-ca` o `verify-full`, también debe configurar el parámetro `connection.ssl.root-cert.location`.
connection.ssl.root-cert.location	Cuando el modo de cifrado de transporte requiere un certificado, especifica la ruta al archivo del certificado.	String	No	Ninguna	Cuando connection.ssl.mode se establece en verify-ca o verify-full, también debe configurar la ruta al certificado de CA. Puede cargar el certificado en la plataforma mediante la función File Management en la consola de Realtime Compute. Una vez cargado, el certificado se almacena en el directorio /flink/usrlib. Por ejemplo, si el archivo del certificado de CA se llama certificate.crt, el valor del parámetro debe ser `'/flink/usrlib/certificate.crt'`. Nota Para obtener un certificado de CA, consulte Transport encryption - Download a CA certificate.
connection.akv4.enabled	Indica si se habilita el modo AKV4 para conectarse al servidor de Hologres.	Boolean	No	false	N/A.
connection.akv4.region	Cuando el modo AKV4 está habilitado, especifica la región donde se encuentra el servidor.	String	No	Ninguna	Por ejemplo, `cn-shanghai`.

Reutilizar un Catalog existente

A partir de VVR 11.5, puede hacer referencia directamente a un Catalog de Hologres integrado creado en la página Data Management dentro de un trabajo de ingesta de datos de Flink CDC. Esto reduce la necesidad de especificar manualmente las propiedades de conexión.

sink:
  type: hologres
  using.built-in-catalog: my_holo_catalog

Los trabajos de ingesta de datos pueden reutilizar automáticamente los siguientes parámetros del Catalog de Hologres:

endpoint
username
password
dbname

Para anular estos parámetros que se reutilizan automáticamente, especifique de forma explícita los parámetros YAML correspondientes, los cuales tienen mayor prioridad.

Mapeo de tipos de datos

Utilice el parámetro sink.type-normalize-strategy para establecer la estrategia que convierte los datos de origen a tipos de Hologres.

Nota

Habilite sink.type-normalize-strategy cuando inicie un trabajo YAML por primera vez. Si lo habilita después de iniciar el trabajo, debe eliminar la tabla de destino y reiniciar el trabajo sin estado para que la configuración surta efecto.
Actualmente, los tipos de arreglo solo admiten INTEGER, BIGINT, FLOAT, DOUBLE, BOOLEAN, CHAR y VARCHAR.
Hologres no admite el tipo numeric como clave principal. Si el tipo de una clave principal se asigna a numeric, el sistema lo convierte al tipo varchar.

STANDARD

Cuando sink.type-normalize-strategy se establece en STANDARD, los tipos se mapean de la siguiente manera:

Tipo de Flink CDC	Tipo de Hologres
CHAR	bpchar
STRING	text
VARCHAR	text (si longitud > 10485760)
VARCHAR	varchar (si longitud <= 10485760)
BOOLEAN	bool
BINARY	bytea
VARBINARY	bytea
DECIMAL	numeric
TINYINT	int2
SMALLINT	int2
INTEGER	int4
BIGINT	int8
FLOAT	float4
DOUBLE	float8
DATE	date
TIME_WITHOUT_TIME_ZONE	time
TIMESTAMP_WITHOUT_TIME_ZONE	timestamp
TIMESTAMP_WITH_LOCAL_TIME_ZONE	timestamptz
ARRAY	Arreglos de los tipos de elemento correspondientes
MAP	No compatible
ROW	No compatible

BROADEN

Cuando sink.type-normalize-strategy se establece en BROADEN, los tipos de Flink CDC se convierten a tipos de Hologres con un rango más amplio. Los tipos se mapean de la siguiente manera:

Tipo de Flink CDC	Tipo de Hologres
CHAR	text
STRING
VARCHAR
VARCHAR
BOOLEAN	bool
BINARY	bytea
VARBINARY	bytea
DECIMAL	numeric
TINYINT	int8
SMALLINT
INTEGER
BIGINT
FLOAT	float8
DOUBLE	float8
DATE	date
TIME_WITHOUT_TIME_ZONE	time
TIMESTAMP_WITHOUT_TIME_ZONE	timestamp
TIMESTAMP_WITH_LOCAL_TIME_ZONE	timestamptz
ARRAY	Arreglos de los tipos de elemento correspondientes
MAP	No compatible
ROW	No compatible

ONLY_BIGINT_OR_TEXT

Cuando sink.type-normalize-strategy se establece en ONLY_BIGINT_OR_TEXT, todos los tipos de Flink CDC se convierten a tipos BIGINT o STRING en Hologres. El mapeo de tipos de datos es el siguiente:

Tipo de Flink CDC	Tipo de Hologres
TINYINT	int8
SMALLINT
INTEGER
BIGINT
BOOLEAN	text
BINARY
VARBINARY
DECIMAL
FLOAT
DOUBLE
DATE
TIME_WITHOUT_TIME_ZONE
TIMESTAMP_WITHOUT_TIME_ZONE
TIMESTAMP_WITH_LOCAL_TIME_ZONE
ARRAY	Arreglos de los tipos de elemento correspondientes
MAP	No compatible
ROW	No compatible

Escritura en tablas particionadas

Puede usar el sink de Hologres junto con una transformación para escribir los datos de origen en una tabla particionada de Hologres.

La partition key debe formar parte de la primary key. Si utiliza una columna que no es clave principal de los datos de origen como partition key, las claves principales de las tablas de origen y destino pueden volverse inconsistentes. Esto provoca discrepancias durante la sincronización de datos.
Hologres admite columnas de los tipos de datos TEXT, VARCHAR y INT como partition key. A partir de la versión 1.3.22, también se admiten columnas del tipo de dato DATE.
Para crear automáticamente las tablas particionadas secundarias, establezca el parámetro createparttable en true. De lo contrario, deberá crearlas manualmente.

Consulte Writing data to a partitioned table para ver un ejemplo.

Sincronización del esquema de tabla

Un pipeline YAML de CDC emplea distintas estrategias para gestionar los cambios en el esquema de la tabla. Estas se configuran mediante el parámetro schema.change.behavior a nivel de pipeline. Los valores válidos para schema.change.behavior son IGNORE, LENIENT, TRY_EVOLVE, EVOLVE y EXCEPTION. Actualmente, el sink de Hologres no admite la estrategia TRY_EVOLVE. Las estrategias LENIENT y EVOLVE implican modificaciones en el esquema de la tabla. En las siguientes secciones se describe cómo estos dos modos gestionan distintos eventos de cambio de esquema.

LENIENT (predeterminado)

En el modo LENIENT, los cambios de esquema se gestionan de la siguiente manera:

Agregar una columna que acepta valores nulos: la columna correspondiente se agrega automáticamente al final de la tabla de destino y sus datos se sincronizan.
Eliminar una columna que acepta valores nulos: la columna no se elimina de la tabla de destino. En su lugar, se llena automáticamente con valores NULL.
Agregar una columna que no acepta valores nulos: se añade automáticamente una columna equivalente que acepta valores nulos al final de la tabla de destino, y sus datos se sincronizan. Para las filas existentes, esta nueva columna se llena de forma automática con NULL.
Cambiar el nombre de una columna: esta operación se trata como la eliminación de una columna y la adición de una nueva. Se agrega una columna con el nombre especificado al final de la tabla de destino, mientras que la columna con el nombre original se llena automáticamente con valores NULL. Por ejemplo, si se cambia el nombre de col_a a col_b, se añade la columna col_b al final de la tabla de destino y la columna col_a se rellena de forma automática con valores NULL.
Cambiar el tipo de dato de una columna: no compatible. Dado que Hologres no admite modificar el tipo de dato de una columna, debe utilizar el parámetro sink.type-normalize-strategy.
Los siguientes cambios de esquema no son compatibles:
- Modificaciones en restricciones, como la clave principal o el índice.
- Eliminación de una columna que no acepta valores nulos.
- Cambio de una columna de NOT NULL a NULLABLE.

EVOLVE

En el modo EVOLVE, los cambios de esquema se gestionan de la siguiente manera:

Agregar una columna que acepta valores nulos: compatible.
Eliminar una columna que acepta valores nulos: no compatible.
Agregar una columna que no acepta valores nulos: se añade una nueva columna que acepta valores nulos a la tabla de destino.
Cambiar el nombre de una columna: compatible. La columna original se renombra en la tabla de destino.
Cambiar el tipo de dato de una columna: no compatible. Dado que Hologres no admite modificar el tipo de dato de una columna, debe utilizar el parámetro sink.type-normalize-strategy.
Los siguientes cambios de esquema no son compatibles:
- Modificaciones en restricciones, como la clave principal o el índice.
- Eliminación de una columna que no acepta valores nulos.
- Cambio de una columna de NOT NULL a NULLABLE.

Aviso

En el modo EVOLVE, reiniciar el trabajo sin estado y sin eliminar la tabla de destino puede provocar que el pipeline falle debido a inconsistencias de esquema entre los datos de origen y la tabla de destino. En ese caso, deberá ajustar manualmente el esquema de la tabla de destino.

Consulte Enabling the EVOLVE mode para ver un ejemplo.

Ejemplos de código

Ampliación de tipos

Utilice el parámetro sink.type-normalize-strategy para configurar la ampliación de tipos.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Escritura en una tabla particionada

Convierta el campo de marca de tiempo create_time a un tipo de fecha y utilícelo como clave de partición para la tabla de Hologres.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true
 
transform:
  - source-table: test_db.test_source_table
    projection: \*, DATE_FORMAT(CAST(create_time AS TIMESTAMP), 'yyyy-MM-dd') as partition_key
    primary-keys: id, create_time, partition_key
    partition-keys: partition_key
    description: add partition key 

pipeline:
  name: MySQL to Hologres Pipeline

Habilitación del modo EVOLVE****

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Automatically create partitioned tables if they do not exist.
  createparttable: true

pipeline:
  name: MySQL to Hologres Pipeline
  schema.change.behavior: evolve

Sincronización de una sola tabla

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.test_source_table
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Sincronización completa de la base de datos

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Fusión de tablas fragmentadas

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.  
  sink.type-normalize-strategy: BROADEN
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Hologres table named test_db.user.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Hologres Pipeline

Sincronización con un esquema específico

En Hologres, un esquema equivale a una base de datos en MySQL. Puede especificar el esquema para las tablas de destino.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN
  
route:
  # Synchronize all tables from the MySQL test_db database to the Hologres test_db2 schema while keeping the original table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Hologres Pipeline

Sincronización de tablas nuevas sin reinicio

Para sincronizar en tiempo real las tablas recién agregadas mientras un trabajo está en ejecución, establezca scan.binlog.newly-added-table.enable = true.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  # Automatically captures new tables created while the job is running.  
  scan.binlog.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Adición de tablas existentes al reiniciar

Para incluir una tabla existente en la sincronización, establezca scan.newly-added-table.enabled en true y reinicie el trabajo.

Aviso

No utilice scan.newly-added-table.enabled en un trabajo que previamente se ejecutó con scan.binlog.newly-added-table.enabled. Esta combinación provoca duplicación de datos al reiniciar.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # On restart, the job scans for new tables matching the `tables` parameter and performs a snapshot.
  # Note: This parameter must be used with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Exclusión de tablas

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Excludes tables that match this regular expression.
  tables.exclude: test_db.table1
  server-id: 5401-5499

sink:
  type: hologres
  name: Hologres Sink
  endpoint: <yourEndpoint>
  dbname: <yourDbname>
  username: ${secret_values.ak_id}
  password: ${secret_values.ak_secret}
  # Map CDC data types to broader Hologres types.
  sink.type-normalize-strategy: BROADEN

pipeline:
  name: MySQL to Hologres Pipeline

Documentos relacionados

Para obtener más información sobre los módulos source, sink, transform y route, consulte Flink CDC Data Ingestion Job Development Reference.
Para desarrollar un trabajo YAML de ingesta de datos, consulte Flink CDC Data Ingestion Job Development.