Connect Flink CDC to a DLF catalog - Data Lake Formation - Centro de documentación de Alibaba Cloud

Flink Change Data Capture (CDC) es una herramienta de ingesta de datos que ofrece Realtime Compute for Apache Flink. Permite sincronizar bases de datos completas desde los orígenes hacia su data lakehouse. En este tema se explica cómo usar Flink CDC para ingerir datos en un catálogo de DLF en tiempo real mediante Paimon REST.

Al finalizar este tema, usted habrá:

Creado un catálogo de DLF como destino de la ingesta.
Configurado un trabajo YAML de Flink CDC con el sink de Paimon apuntando a DLF.
Ejecutado ejemplos para escenarios comunes de ingesta: sincronización completa de base de datos, tablas particionadas, tablas de solo anexado y orígenes de Kafka.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Un espacio de trabajo de Realtime Compute for Apache Flink. Para más detalles, consulte Crear un espacio de trabajo.
El espacio de trabajo de Realtime Compute for Apache Flink y los catálogos de DLF en la misma región.
La VPC de su espacio de trabajo de Realtime Compute for Apache Flink agregada a la lista blanca de VPC de DLF. Para más información, consulte Configurar una lista blanca de VPC.

Requisitos del motor

Su trabajo de Realtime Compute for Apache Flink debe ejecutar la versión 11.1.0 o posterior de Ververica Runtime (VVR).

Crear un catálogo de DLF

Consulte Primeros pasos con DLF.

Crear y configurar un trabajo de ingesta de datos

Cree un borrador YAML para ingerir datos con Flink CDC. Para más información, consulte Develop Flink CDC jobs for data ingestion (Beta).

Configure el módulo sink:

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Reemplace los valores de marcador de posición con sus valores reales:

Opción de configuración	Descripción	Obligatorio	Predeterminado	Valor de ejemplo
`catalog.properties.metastore`	El tipo de metastore. Establezca el valor `rest`.	Sí	—	`rest`
`catalog.properties.token.provider`	El proveedor de token. Establezca el valor `dlf`.	Sí	—	`dlf`
`catalog.properties.uri`	La URI para acceder al servidor del catálogo REST de DLF. Formato: `http://[region-id]-vpc.dlf.aliyuncs.com`. Para conocer los identificadores de región, consulte Regions and endpoints.	Sí	—	`http://ap-southeast-1-vpc.dlf.aliyuncs.com`
`catalog.properties.warehouse`	El nombre del catálogo de Paimon.	Sí	—	`dlf_test`
`commit.user`	El usuario de confirmación para las escrituras de datos. Asigne usuarios de confirmación únicos a trabajos distintos para evitar conflictos.	No	`admin`	`your_job_name`
`table.properties.deletion-vectors.enabled`	Habilita los vectores de eliminación para mejorar el rendimiento de lectura con un impacto mínimo en las escrituras.	No	—	`true`

Notas de uso

Antes de ejecutar un trabajo, tenga en cuenta las siguientes restricciones:

Conflictos de usuario de confirmación: el usuario de confirmación predeterminado es admin. Si ejecuta varios trabajos simultáneos de escritura de datos en la misma tabla con el mismo usuario de confirmación, se generan conflictos e inconsistencias en los datos. Asigne un valor único de commit.user a cada trabajo.
Sin opciones de compactación ni de bucket: DLF proporciona compactación automática de archivos. No configure opciones de compactación ni de bucket, como bucket y num-sorted-run.compaction-trigger.
Vectores de eliminación: establezca table.properties.deletion-vectors.enabled: true para acelerar considerablemente las lecturas con un impacto mínimo en las escrituras.

Ejemplos

Ingerir datos de una base de datos MySQL completa a DLF

A continuación se muestra un trabajo que sincroniza todas las tablas de una base de datos MySQL con DLF:

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Las opciones recomendadas para el origen de MySQL son las siguientes:

Opción	Descripción
`scan.binlog.newly-added-table.enabled`	Sincroniza los datos de las tablas creadas durante la fase incremental.
`include-comments.enabled`	Sincroniza los comentarios de tablas y campos.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Previene posibles errores OOM en TaskManager al priorizar los fragmentos no acotados.
`scan.only.deserialize.captured.tables.changelog.enabled`	Deserializa los datos únicamente de las tablas coincidentes para acelerar las lecturas.

Ingerir datos en una tabla particionada

Para ingerir datos de una tabla de origen sin particiones a una tabla particionada en DLF, agregue la opción partition-keys al módulo transform. Para más información, consulte Data ingestion with Flink CDC.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
  - source-table: mysql_test.tbl2
    partition-keys: id,pt

Ingerir datos en una tabla de solo anexado

Para implementar eliminaciones lógicas (soft delete) durante la ingesta, utilice converter-after-transform: SOFT_DELETE en el módulo transform. Esta configuración convierte las operaciones de eliminación en inserciones, por lo que la tabla descendente registra todos los cambios de forma completa. El campo __data_event_type__ dentro de projection escribe el tipo de cambio como una nueva columna en la tabla descendente.

source:
  type: mysql
  name: MySQL Source
  hostname: ${mysql.hostname}
  port: ${mysql.port}
  username: ${mysql.username}
  password: ${mysql.password}
  tables: mysql_test.\.*
  server-id: 8601-8604
  # (Optional) Sync data from tables created in the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  # (Optional) Sync table and field comments.
  include-comments.enabled: true
  # (Optional) Prioritize unbounded shards to prevent potential TaskManager out-of-memory (OOM) errors.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  # (Optional) Deserialize data only from matched tables to speed up reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: mysql_test.tbl1
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE
  - source-table: mysql_test.tbl2
    partition-keys: id,pt
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE

Para más información, consulte Data ingestion with Flink CDC.

Sincronizar datos de Kafka a DLF en tiempo real

El siguiente trabajo lee datos CDC del tema inventory de Kafka (tablas customers y products en formato Debezium JSON) y los sincroniza con las tablas de destino correspondientes en DLF. Como los mensajes Debezium JSON no contienen información de clave primaria, esta se especifica explícitamente en el módulo transform.

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: ${kafka.bootstrap.servers}
  topic: inventory
  scan.startup.mode: earliest-offset
  value.format: debezium-json
  debezium-json.distributed-tables: true

sink:
  type: paimon
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  # (Optional) The commit user. Set different commit users for different jobs to avoid conflicts.
  commit.user: your_job_name
  # (Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

# Debezium JSON does not contain primary key info, so explicitly specify primary keys.
transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id

Notas adicionales para los orígenes de Kafka:

El origen de Kafka admite tres formatos de datos: canal-json, debezium-json (predeterminado) y json.
Cuando ingiera datos de varias particiones de Kafka a una sola tabla en DLF, establezca debezium-json.distributed-tables o canal-json.distributed-tables en true.
El origen de Kafka admite varias políticas de inferencia de esquema mediante la opción schema.inference.strategy. Para más información, consulte Message Queue for Apache Kafka.

Para más información, consulte Data ingestion with Flink CDC.