Flink CDC Transform module - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Este tema describe las reglas de sintaxis y las funciones integradas del módulo Transform para los trabajos de ingestión de datos de Flink CDC.

Parámetros de reglas de transformación

El módulo Transform permite la manipulación directa de columnas de datos. Puede eliminar o ampliar columnas existentes y filtrar datos no deseados durante la sincronización. Utilice los siguientes parámetros para definir las reglas de transformación:

Parámetro	Descripción	Obligatorio	Notas
`source-table`	Especifica la tabla ascendente que se va a transformar.	Sí	Admite expresiones regulares.
`projection`	Especifica la regla de proyección para la tabla ascendente. Esto determina todas las columnas de datos después de la transformación.	No	La sintaxis es similar a la instrucción SELECT de SQL. Si no se especifica este parámetro, no se agregan ni eliminan columnas. Para obtener más información, consulte Data screening. Para conocer las funciones integradas disponibles, consulte Flink CDC built-in functions.
`filter`	Regla para el filtrado de filas.	No	La sintaxis es similar a la instrucción WHERE de SQL. Si no se especifica este parámetro, no se filtran filas.
`primary-keys`	Especifica las columnas de clave primaria después de la transformación.	No	Si no se especifica este parámetro, se conserva la definición de clave primaria del esquema original. Las claves primarias se separan con comas (`,`). Importante De forma predeterminada, no puede eliminar columnas de clave primaria del origen. Agregue el parámetro `transform.allow.trimming.pk-columns` a la configuración de `pipeline:` para reducir el número de columnas de clave primaria ascendentes. Al definir columnas de clave primaria personalizadas, asegúrese de que los datos entrantes de la tabla ascendente cumplan con la restricción PRIMARY KEY. Se recomienda incluir las columnas de clave primaria de la tabla ascendente en las columnas de clave primaria personalizadas para evitar problemas de desorden de datos durante escrituras entre particiones.
`partition-keys`	Especifica la lista de columnas de clave de partición después de la transformación.	No	Si no se especifica este parámetro, se conserva la definición de clave de partición del esquema original. Las claves de partición se separan con comas (`,`). Importante Al definir columnas de clave de partición personalizadas, asegúrese de que los datos entrantes de la tabla ascendente cumplan con la restricción PRIMARY KEY para evitar problemas de desorden de datos durante escrituras entre particiones.
`table-options`	Pasa información de configuración adicional al destino.	No	Una lista de propiedades opcionales, como el número de buckets para un destino Paimon o comentarios. Los elementos de configuración se separan con comas (`,`). La clave y el valor de cada elemento se separan con un signo igual (`=`). Ejemplo de configuración: `key1=value1,key2=value2`
`description`	La descripción de la regla de transformación.	No	Ninguna.
`converter-after-transform`	Un convertidor que realiza procesamiento adicional en los datos una vez completada la transformación.	No	Para obtener más información, consulte Converter after Transform.

Notas

Después de modificar una instrucción en el módulo Transform, debe reiniciar el trabajo desde un estado inicial sin estado.
Por lo general, las instrucciones projection y filter no necesitan ir entre comillas.
```
transform:
  - projection: a, b, c
    # Equivalent to
  - projection: "a, b, c"
```
Sin embargo, si una expresión de proyección comienza con un carácter especial, como * o ', es posible que la expresión completa no se analice como un literal de cadena YAML válido. En este caso, debe encerrar manualmente toda la expresión entre comillas simples (') o dobles ("), o usar una barra invertida (\) para escapar el carácter.
```
transform:
  - projection: *, 42      # Not valid YAML
  - projection: '*, 42'    # OK
  - projection: \*, 42     # OK  
```
A partir de Ververica Runtime (VVR) 11.6, puede purgar la configuración de clave primaria inferida del origen y convertir la tabla en una sin clave primaria. La sintaxis es la siguiente:
```
transform:
  - source-table: db.tbl
    primary-keys: '' # Clears the PRIMARY KEY setting for the db.tbl table
```

Selección de datos

El módulo Transform utiliza una sintaxis similar a SQL para definir reglas de selección de datos (proyección). Estas reglas le permiten seleccionar columnas específicas, agregar columnas calculadas y añadir columnas de metadatos.

Reducción de columnas

Para extraer columnas específicas de la tabla de origen y sincronizarlas con el destino, puede especificar las columnas deseadas en la regla de proyección. Las columnas que no se especifiquen no se envían al destino.

transform:
  - source-table: db.tbl
    projection: col_1, col_3, col_4 # col_2 is pruned

Importante

La reducción de columnas puede provocar inconsistencias entre los esquemas de las tablas ascendente y descendente si cambia el esquema de la tabla ascendente.

Caracteres comodín

Para enviar todas las columnas de la tabla de origen al destino sin modificaciones, incluidas las nuevas columnas que se agreguen posteriormente, puede utilizar el carácter comodín asterisco (*) en la regla de proyección.

Nota

Si una regla de proyección no utiliza un carácter comodín (*), el esquema resultante es fijo. El esquema permanece consistente con la versión especificada en la regla de proyección y la evolución del esquema no se puede sincronizar.

Por ejemplo, *, 'extras' AS extras indica que se agrega una columna adicional después de las columnas del esquema ascendente. La evolución del esquema ascendente todavía se sincroniza con el destino.

transform:
  - source-table: db.tbl
    projection: \*, 'extras' AS extras

Columnas calculadas

Puede usar la sintaxis <Expression> AS <ColName> en una regla de proyección para agregar una columna calculada. La expresión se evalúa para cada registro del origen y el resultado llena la columna correspondiente.

Nota

La expresión de una columna calculada no puede hacer referencia al valor de otra columna calculada, incluso si la columna referenciada aparece antes. Por ejemplo, a, b AS c, c AS d no es una expresión válida.

Por ejemplo, cuando la tabla ascendente db.tbl envía el registro de datos [+I, id = 1], el registro se transforma en la fila de datos [+I, id = 1, inc_id = 2] y luego se envía al destino.

transform:
  - source-table: db.tbl
    projection: id, id + 1 AS inc_id

Columnas de metadatos

Al escribir reglas de proyección, puede usar las siguientes columnas de metadatos predefinidas de la misma manera que las columnas de datos normales:

Importante

No defina columnas de datos normales con los mismos nombres que las columnas de metadatos.

Las siguientes columnas de metadatos se aplican a todos los conectores.

Nombre de columna de metadatos	Tipo de dato	Descripción
`__namespace_name__`	String	El nombre del namespace de la tabla de origen para este registro de cambio.
`__schema_name__`	String	El nombre del esquema de la tabla de origen para este registro de cambio.
`__table_name__`	String	El nombre de la tabla de origen para este registro de cambio.
`__data_event_type__`	String	El tipo de operación para este registro de cambio (`+I`, `-U`, `+U` o `-D`). Importante Un evento CDC siempre empaqueta los registros Update Before y Update After de una actualización en un solo evento. Por lo tanto, el contenido de `__data_event_type__` es `-U` y `+U` respectivamente dentro del mismo evento de actualización. No lo utilice como clave primaria.

Por ejemplo, puede escribir el nombre completo de la tabla ascendente en una columna calculada y enviarlo al destino.

transform:
  - source-table: \.*.\.*
    projection: \*, __namespace_name__ || __schema_name__ || __table_name__ AS identifier

La siguiente tabla muestra las relaciones de asignación entre los conectores de base de datos y los nombres de Namespace, Schema y Table.

Tipo de base de datos	Nombre de namespace	Nombre de esquema	Nombre de tabla
MySQL	-	Database	Table
Kafka	-	-	Topic
SLS	-	Project	LogStore
MongoDB	-	Database	Collection
Paimon	-	Database	Table
Hologres	-	Schema	Table
StarRocks	-	Database	Table
Doris	-	Database	Table
Postgres	Database Nota Esto solo tiene efecto cuando el parámetro `table-id.include-database` está habilitado.	Schema	Table

Filtrado de datos

El módulo Transform utiliza una sintaxis similar a SQL para definir reglas de filtrado de filas.

La regla de filtro debe ser una expresión que se evalúe como un valor BOOLEAN. Puede hacer referencia a cualquier columna de la tabla de origen y a cualquier columna calculada.

Si un registro de cambio coincide con una regla de transformación que tiene un filtro, y la expresión del filtro se evalúa como FALSE, la fila de datos no se envía al destino.

Nota

Si utiliza una columna calculada en una regla de proyección para sobrescribir una columna ascendente existente, la expresión de filtro hace referencia a la columna calculada.

Por ejemplo, la siguiente regla de transformación es válida:

transform:
  - source-table: db.tbl
    projection: CAST(id AS VARCHAR) AS id
    filter: CHAR_LENGTH(id) > 5

La columna id a la que se hace referencia en la expresión filter es la columna calculada que se convirtió al tipo VARCHAR.

Coincidencia de patrones de cadena

Utilice LIKE y NOT LIKE en expresiones de filtro para realizar coincidencias de patrones de cadena. La implementación utiliza la semántica find() de expresiones regulares de Java, la cual difiere del estándar SQL LIKE en dos aspectos importantes:

• El patrón es una expresión regular de Java. % y _ se tratan como literales de regex en lugar de comodines.

• La coincidencia utiliza semántica de subcadena. El patrón solo necesita aparecer en alguna posición de la cadena para devolver TRUE; no se requiere una coincidencia de cadena completa.

Ejemplo: Filtrar registros cuya columna code comience con HSRS.

El siguiente patrón coincide incorrectamente — ABC_HSRS123 también coincide:

transform:
  - source-table: db.orders
    filter: code LIKE 'HSRS.*'

El patrón correcto utiliza el ancla de inicio ^:

transform:
  - source-table: db.orders
    filter: code LIKE '^HSRS'

Los equivalentes de patrones comunes son los siguientes:

Sintaxis SQL	Regex CDC equivalente	Significado
`col LIKE 'HSRS%'`	`col LIKE '^HSRS'`	Comienza con HSRS
`col LIKE '%HSRS'`	`col LIKE 'HSRS$'`	Termina con HSRS
`col LIKE '%HSRS%'`	`col LIKE 'HSRS'`	Contiene HSRS
`col LIKE 'HSRS'`	`col LIKE '^HSRS$'`	Es exactamente igual a HSRS

Para consultar la semántica completa de LIKE y más equivalentes de regex, consulte Flink CDC built-in functions.

Reglas de configuración avanzada

Referencia a columnas no reducidas y metadatos

Ejemplo 1: Filtrar basado en una columna reducida

Suponga que el esquema de la tabla ascendente es [INT a, INT b, BOOLEAN c]. Para generar las columnas a y b y conservar solo las filas donde c es true, puede usar la siguiente regla de configuración:

transform:
  - source-table: ...
    projection: a, b
    filter: c

Ejemplo 2: Filtrar basado en columnas de metadatos

Puede usar columnas de metadatos directamente como condiciones de filtro sin definirlas explícitamente en projection. Por ejemplo, puede filtrar datos de cambio del tipo DELETE.

transform:
  - source-table: db.tbl
    projection: a, b
    filter: __data_event_type__ <> '-D'

Sobrescribir columnas existentes

En projection, puede definir un campo con el mismo nombre que una columna ascendente para sobrescribir el valor o el tipo de esa columna. Este método asegura que la evolución del esquema se sincronice correctamente.

Ejemplo: Forzar conversión de tipo

Suponga que el esquema de la tabla ascendente es [INT a, INT b, BOOLEAN c]. Para mantener los nombres de columna sin cambios pero forzar la conversión de la columna a a un tipo de cadena, puede usar la siguiente configuración:

transform:
  - source-table: db.tbl
    projection: \*, CAST(a AS VARCHAR) AS a

El esquema de la tabla descendente se convierte en [STRING a, INT b, BOOLEAN c]. La columna original a se sobrescribe con la nueva definición.

Reutilizar columnas calculadas en condiciones de filtro

Un filter puede hacer referencia directamente al alias de una columna calculada definida en una projection.

Ejemplo: Referenciar un resultado calculado

Puede definir una nueva columna d en projection y usarla directamente en filter.

transform:
  - source-table: db.tbl
    projection: a, b, c, a + b + c AS d
    filter: d > 100

Esta configuración tiene el mismo efecto que la siguiente, pero es más legible:

transform:
  - source-table: ...
    projection: a, b, c, a + b + c AS d
    filter: a + b + c > 100

Converter after Transform

converter-after-transform se utiliza para procesar cambios en los datos después de aplicar todas las reglas de transformación. Puede especificar varios convertidores separándolos con comas (,). Los convertidores se aplican en el orden en que aparecen listados. Se admiten los siguientes valores de configuración.

Nombre del convertidor	Función	Versiones compatibles
SOFT_DELETE	Convierte cambios de eliminación en inserciones.	VVR 8.0.11 y posteriores.
FIELD_NAME_LOWER_CASE	Convierte todos los nombres de campos de tabla a minúsculas.	VVR 11.1 y posteriores.

Eliminación lógica

El convertidor SOFT_DELETE funciona con la columna de metadatos __data_event_type__ para implementar eliminaciones lógicas. Esto significa que los datos no se eliminan físicamente en el sistema descendente. Por ejemplo, la siguiente configuración de transformación implementa una eliminación lógica. Una operación de eliminación se convierte en una inserción, y el op_type de los datos correspondientes se actualiza a -D para indicar la eliminación.

transform: 
  - source-table: db.tbl
    projection: \*, __data_event_type__ AS op_type
    converter-after-transform: SOFT_DELETE