Módulos de origen y destino - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Este tema describe los módulos de origen y destino para los trabajos de ingestión de datos de Flink Change Data Capture (CDC) y enumera los conectores compatibles.

Conectores compatibles

Conector	Tipo compatible
Conector	Origen	Destino
MySQL Nota Admite conexiones a ApsaraDB RDS for MySQL, PolarDB for MySQL y MySQL autogestionado.	√	×
Paimon	×	√
Kafka	√ Nota Compatible solo con VVR 8.0.10 y versiones posteriores.	√
Upsert Kafka	×	√
StarRocks	×	√
Hologres	×	√
Simple Log Service	√ Nota Compatible solo con VVR 11.1 y versiones posteriores.	×
MongoDB	√ Nota Compatible solo con VVR 11.2 y versiones posteriores.	×
MaxCompute	×	√ Nota Compatible solo con VVR 11.1 y versiones posteriores.
SelectDB	×	√ Nota Compatible solo con VVR 11.1 y versiones posteriores.
PostgreSQL CDC (Public Preview)	√ Nota Compatible solo con VVR 11.4 y versiones posteriores.	×
Print	×	√
Iceberg	×	√ Nota Compatible solo con VVR 11.6 y versiones posteriores.

Configuración del conector

Configure los parámetros de los conectores de origen y destino en un trabajo de ingestión de datos de Flink CDC. Para obtener detalles sobre los conectores compatibles y sus parámetros, consulte las siguientes secciones.

# Source module
source:
  type: mysql # Or another connector identifier
  name: MySQL Source
  # Other parameters. Use key: value pairs.

# Sink module
sink:
  type: paimon # Or another connector identifier
  name: Paimon Sink
  # Other parameters. Use key: value pairs.

Configuración general

Parámetro	Descripción	Obligatorio	Tipo	Predeterminado	Observaciones
type	Tipo de conector para el origen o el destino.	Sí	String	Ninguna	Ninguna
name	Nombre del nodo.	No	String	Ninguna	Ninguna
using.built-in-catalog	Reutiliza un catálogo integrado.	No	String	Ninguna	Disponible en VVR 11.5 y versiones posteriores. Este parámetro solo es compatible con conectores específicos. Para consultar la lista de conectores compatibles, véase Reuse connection information from a catalog.

Reutilizar información de conexión de un catálogo

A partir de VVR 11.5, puede hacer referencia directa a un catálogo integrado creado en la página Data Management para su trabajo de ingestión de datos de Flink CDC. Esto permite reutilizar propiedades de conexión como URL, nombre de usuario y contraseña, lo que simplifica la configuración manual.

Sintaxis

source:
  type: mysql
  using.built-in-catalog: mysql_rds_catalog
  
sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog

Utilice el parámetro using.built-in-catalog en los módulos source y sink para hacer referencia a un catálogo integrado existente.

Por ejemplo, en el caso anterior, los metadatos del catálogo para mysql_rds_catalog ya incluyen parámetros obligatorios como hostname, username y password. Por lo tanto, no es necesario proporcionar estos parámetros nuevamente en el trabajo YAML.

Notas de uso

Los siguientes conectores admiten la reutilización de información de conexión desde un catálogo:

MySQL (origen)
Kafka (origen)
Upsert Kafka (destino)
StarRocks (destino)
Hologres (destino)
Paimon (destino)
Simple Log Service (origen)
Iceberg (destino)

Nota

Los parámetros del catálogo incompatibles con la configuración YAML de CDC se ignoran. Para obtener más información, consulte la lista de parámetros de cada conector.

Configuración de origen

Parámetro	Descripción	Obligatorio	Tipo	Predeterminado	Observaciones
source-expand	Especifica la estrategia de distribución para los datos emitidos desde el origen.	No	Consulte la sección de sintaxis para obtener detalles de configuración.	Ninguna	Disponible en VVR 11.6 y versiones posteriores.

source-expand

El parámetro source-expand distribuye o replica datos antes de que los módulos Transform y Route los procesen.

Sintaxis

source:                                                                                                                                                                                                                                                                                                          
  type: mysql                                                                                                                                                                                                                                                                                                     
  host: localhost                                                                                                                                                                                                                                                                                                  
  port: 3306                                                                                                                                                                                                                                                                                                       
  username: admin
  password: pass
  tables: mydb.orders
  source-expand:
  # Replicate mydb.orders into three tables, each processed differently before writing to the sink.
    - input-table: mydb.orders
      output-table: [ dwd.orders_full, dws.orders_summary, ads.orders_report ]

# Apply different transformations to each of the three expanded tables
transform:
  # DWD layer: Retain all fields, add a calculated column
  - source-table: dwd.orders_full
    projection: "*, amount * discount as final_price"
  # DWS layer: Retain only key fields, filter out small orders
  - source-table: dws.orders_summary
    projection: order_id, user_id, amount, order_status
    filter: amount > 100
    primary-keys: order_id
  # ADS layer: Retain only fields needed for reporting, filter out canceled orders
  - source-table: ads.orders_report
    projection: order_id, user_id, amount, TO_UPPER(order_status) as status
    filter: order_status <> 'CANCELLED'
    primary-keys: order_id

# Route the three expanded tables to different sink tables
route:
  - source-table: dwd.orders_full
    sink-table: starrocks_dwd.orders_full_detail
  - source-table: dws.orders_summary
    sink-table: starrocks_dws.orders_summary
  - source-table: ads.orders_report
    sink-table: starrocks_ads.orders_report

sink:
  type: starrocks
  name: sink-starrocks
  jdbc-url: jdbc:mysql://localhost:9030
  load-url: localhost:8030
  username: root
  password: pass

Como se muestra en el ejemplo, esta configuración replica los datos de la tabla mydb.orders en tres tablas lógicas: dwd.orders_full, dws.orders_summary y ads.orders_report. Posteriormente, cada una de estas tablas se procesa de forma distinta antes de escribirse en una tabla de destino separada.

Notas de uso

Disponible en VVR 11.6 y versiones posteriores.
De forma predeterminada, la tabla original no se conserva después de la distribución de datos. En el ejemplo de sintaxis, la tabla mydb.orders se elimina del flujo de datos. Para conservar la tabla original, agréguela a la lista output-table.
```
source-expand:
  - input-table: mydb.orders
    output-table: [ mydb.orders, db1.orders, db2.orders ]
```
Los parámetros input-table y output-table no admiten expresiones regulares.

Configuración de destino

Parámetro	Descripción	Obligatorio	Tipo	Predeterminado	Observaciones
include.schema.changes	Especifica qué tipos de cambios de esquema aplicar.	No	List<String>	Ninguna	De forma predeterminada, se sincronizan todos los cambios de esquema.
exclude.schema.changes	Especifica qué tipos de cambios de esquema excluir.	No	List<String>	Ninguna	Tiene mayor prioridad que `include.schema.changes`.

Para obtener más información sobre cómo usar include.schema.changes y exclude.schema.changes, consulte Schema Change Synchronization Configuration.