Flink CDC job structure - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Un trabajo de ingestión de datos de Flink CDC se define como un archivo YAML compuesto por hasta cinco módulos. Los módulos Source y Sink son obligatorios; los módulos Transform, Route y Pipeline son opcionales.

Descripción general de la estructura del trabajo

Un archivo de trabajo completo sigue esta estructura:

source:         # Required: defines the data source
  ...

sink:           # Required: defines the destination
  ...

transform:      # Optional: filters and reshapes data before writing
  - ...

route:          # Optional: maps source tables to different sink tables
  - ...

pipeline:       # Optional: sets global job parameters
  ...

Configuración mínima

El siguiente ejemplo sincroniza todas las tablas coincidentes de MySQL a Paimon mediante únicamente los módulos obligatorios:

# Source module
source:
  type: mysql
  name: MySQL Source
  host: localhost
  port: 3306
  username: admin
  password: <yourPassword>
  tables: adb.*

# Sink module
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

# Pipeline module
pipeline:
  name: source-database-sync-pipe
  schema.change.behavior: evolve

Configuración completa

El siguiente ejemplo agrega los módulos Transform y Route para filtrar campos, aplicar expresiones y reasignar tablas:

# Source module
source:
  type: mysql
  name: MySQL Source
  host: localhost
  port: 3306
  username: admin
  password: <yourPassword>
  tables: adb.*, bdb.user_table_[0-9]+, [app|web]_order_.*, mydb.\\.*

# Sink module
sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: /path/warehouse

# Transform module
transform:
  - source-table: mydb.app_order_.*
    projection: id, order_id, TO_UPPER(product_name)
    filter: id > 10 AND order_id > 100
    primary-keys: id
    partition-keys: product_name
    table-options: comment=app order
    description: project fields from source table
    converter-after-transform: SOFT_DELETE
  - source-table: mydb.web_order_.*
    projection: CONCAT(id, order_id) as uniq_id, *
    filter: uniq_id > 10
    description: add new uniq_id for each row

# Route module
route:
  - source-table: mydb.default.app_order_.*
    sink-table: odsdb.default.app_order
    description: sync all table shards to one
  - source-table: mydb.default.web_order
    sink-table: odsdb.default.ods_web_order
    description: sync table with given prefix ods_

# Pipeline module
pipeline:
  name: source-database-sync-pipe
  schema.change.behavior: evolve

Módulo Source

El módulo Source define el origen desde el cual Flink CDC lee los datos. Conectores compatibles:

source:
  type: mysql      # Connector type
  name: mysql source
  # Connector-specific parameters follow
  xxx: ...

Para consultar todos los parámetros compatibles, consulte la documentación del conector correspondiente.

Módulo Sink

El módulo Sink define el destino donde Flink CDC escribe los datos. Conectores compatibles:

sink:
  type: hologres   # Connector type
  name: hologres sink
  # Connector-specific parameters follow
  xxx: ...

Para consultar todos los parámetros compatibles, consulte la documentación del conector correspondiente.

Módulo Transform

El módulo Transform aplica reglas de proyección, cálculo y filtrado a los datos de la tabla de origen antes de que lleguen al destino. Defina una regla por cada patrón de tabla de origen.

transform:
  - source-table: db.tbl1
    projection: ...
    filter: ...
  - source-table: db.tbl2
    projection: ...
    filter: ...

Parámetro	Descripción	Obligatorio
`source-table`	Identificador de la tabla de origen; admite expresiones regulares	Obligatorio
`projection`	Selección de columnas y expresiones, similar a la cláusula `SELECT` de SQL	Opcional
`filter`	Condición de filtrado de filas, similar a la cláusula `WHERE` de SQL	Opcional
`primary-keys`	Claves principales de la tabla de destino, separadas por comas	Opcional
`partition-keys`	Claves de partición de la tabla de destino, separadas por comas	Opcional
`table-options`	Opciones de creación de tablas que se aplican cuando la tabla de destino se crea automáticamente	Opcional
`converter-after-transform`	Convertidor que se aplica a los eventos de cambio después de ejecutar la regla de transformación	Opcional
`description`	Descripción de la regla	Opcional

Para obtener detalles sobre la sintaxis de las expresiones, consulte Flink CDC Transform module.

Módulo Route

El módulo Route asigna tablas de origen a tablas de destino. Utilícelo para fusionar tablas fragmentadas, agregar prefijos a los nombres de las tablas o redirigir tablas a diferentes destinos.

route:
  - source-table: db.tbl1
    sink-table: sinkdb.tbl1
  - source-table: db.tbl2
    sink-table: sinkdb.tbl2

Parámetro	Descripción	Obligatorio
`source-table`	Identificador de la tabla de origen; admite expresiones regulares	Obligatorio
`sink-table`	Identificador de la tabla de destino	Obligatorio
`description`	Descripción de la regla de enrutamiento	Opcional

Para obtener detalles sobre la sintaxis de las expresiones, consulte Flink CDC Route module.

Módulo Pipeline

El módulo Pipeline establece parámetros globales que se aplican a todo el trabajo.

pipeline:
  name: CDC YAML job
  schema.change.behavior: LENIENT

Parámetro	Descripción	Obligatorio
`name`	Nombre del trabajo que se muestra en el panel de Flink	Opcional
`schema.change.behavior`	Define cómo gestiona el trabajo los cambios de esquema en el origen. Valores aceptados: `evolve`, `LENIENT`, entre otros.	Opcional

Para consultar todos los parámetros configurables, consulte Flink CDC Pipeline module.

Realtime Compute for Apache Flink:Flink CDC job structure