Paimon connector - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

Este tema explica cómo usar el conector de Paimon en un data lakehouse de streaming. Para obtener mejores resultados, se recomienda utilizar este conector junto con Paimon Catalog.

Contexto

Apache Paimon es un formato de almacenamiento de data lake unificado para datos de streaming y por lotes que admite escrituras de alto rendimiento y consultas de baja latencia. Se integra de manera óptima con los motores de cómputo más populares disponibles en Alibaba Cloud E-MapReduce, como Flink, Spark, Hive y Trino. Apache Paimon le permite construir rápidamente un data lake en HDFS u OSS y conectarlo a estos motores de cómputo para analizar los datos. Para obtener más información, consulte Apache Paimon.

Categoría	Descripción
Tipos compatibles	Tabla de origen, tabla de dimensiones, tabla de resultados y destino para la ingesta de datos
Modo de ejecución	Modo de streaming y modo por lotes
Formato de datos	No compatible
Métricas de monitoreo	Ninguna
Tipos de API	SQL y YAML para la ingesta de datos
Actualizaciones y eliminaciones en tablas de resultados	Sí

Funciones principales

Apache Paimon ofrece las siguientes capacidades principales:

Construir un data lake ligero y de bajo costo en HDFS o en object storage.
Leer y escribir conjuntos de datos a gran escala, tanto en modo de streaming como por lotes.
Ejecutar consultas por lotes y OLAP con una actualización de datos que va desde minutos hasta segundos.
Ingerir y producir datos incrementales, lo cual sirve como capa de almacenamiento para almacenes de datos tradicionales sin conexión y modernos de streaming.
Preagregar datos para reducir los costos de almacenamiento y la carga de cómputo posterior.
Acceder a versiones históricas de los datos.
Filtrar datos de forma eficiente.
Habilitar la evolución del esquema.

Límites y recomendaciones

El conector de Paimon requiere el motor de cómputo Flink VVR 6.0.6 o una versión posterior.

A continuación se muestra la compatibilidad de versiones entre Paimon y VVR.

Versión de Apache Paimon	VVR
1.3.1	11.5, 11.6, 11.7, 11.8
1.3	11.4
1.2	11.2, 11.3
1.1	11.1
1.0	8.0.11

Recomendaciones de almacenamiento para escrituras concurrentes

Cuando varios trabajos realizan escrituras concurrentes en la misma tabla de Paimon, el uso del almacenamiento estándar de OSS (oss://) puede provocar conflictos de confirmación o trabajos con errores debido a las limitaciones en las operaciones atómicas de archivos.

Para lograr escrituras estables y consistentes, utilice un servicio de metadatos o de almacenamiento que ofrezca garantías atómicas sólidas. La opción preferida es Data Lake Formation (DLF), ya que proporciona una gestión unificada de los metadatos y del almacenamiento de Paimon. Como alternativa, puede usar OSS-HDFS o HDFS.
Aplicación de los cambios de configuración

Los cambios en los parámetros de configuración de una tabla de Paimon surten efecto únicamente después de reiniciar los trabajos relacionados. Los trabajos en ejecución no cargan estos cambios de forma dinámica.
Retraso en la recuperación física de particiones eliminadas

Al ejecutar una operación DROP PARTITION, el sistema no elimina de inmediato los archivos de datos físicos subyacentes. Esta operación realiza una eliminación lógica. Paimon quita los metadatos de la partición de destino solo de la instantánea más reciente. Dado que Paimon es compatible con la función time travel, las instantáneas históricas siguen haciendo referencia a los archivos de datos de esa partición. Los archivos de datos físicos se eliminan de forma permanente únicamente cuando todas las instantáneas históricas que hacen referencia a la partición alcanzan su límite de retención y el mecanismo de expiración de instantáneas las limpia.

SQL

Utilice el conector de Paimon en trabajos SQL como tabla de origen o como tabla de resultados.

Sintaxis

Si crea una tabla de Paimon en un Paimon Catalog, no necesita especificar el parámetro connector. La sintaxis es la siguiente:
```
CREATE TABLE `<YOUR-PAIMON-CATALOG>`.`<YOUR-DB>`.paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  ...
);
```
Nota
Si ya creó una tabla de Paimon en un Paimon Catalog, puede usarla directamente.
Si crea una tabla temporal de Paimon en otro Catalog, debe especificar los parámetros 'connector' y 'path'. La sintaxis es la siguiente:
```
CREATE TEMPORARY TABLE paimon_table (
  id BIGINT,
  data STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'paimon',
  'path' = '<path-to-paimon-table-files>',
  'auto-create' = 'true', -- If Paimon table data files do not exist at the specified path, they are automatically created.
  ...
);
```
Nota
- Ejemplo de ruta: 'path' = 'oss://<bucket>/test/order.db/orders'. No omita el sufijo .db. Paimon depende de este sufijo para identificar la base de datos.
- Los distintos trabajos que escriben en la misma tabla deben compartir la misma configuración de ruta.
- Si dos configuraciones de ruta son diferentes, Paimon no las reconoce como la misma tabla. Incluso si la ruta física es idéntica, las configuraciones inconsistentes del Catalog pueden generar conflictos de escritura simultánea, operaciones compact con errores y pérdida de datos. Por ejemplo, Paimon considera que oss://b/test y oss://b/test/ son tablas distintas debido a la barra diagonal final, aunque apunten a la misma ubicación física.

Parámetros WITH

Parámetro	Descripción	Tipo	Obligatorio	Predeterminado	Observaciones
connector	Especifica el conector de la tabla.	String	No	Ninguna	Este parámetro no es obligatorio si crea una tabla de Paimon en un catálogo de Paimon. Si crea una tabla temporal de Paimon en otro catálogo, este parámetro debe establecerse en `paimon`.
path	Ruta de almacenamiento de la tabla.	String	No	Ninguna	Este parámetro no es obligatorio si crea una tabla de Paimon en un catálogo de Paimon. Si crea una tabla temporal de Paimon en otro catálogo, este parámetro especifica el directorio de almacenamiento de la tabla en HDFS u OSS.
auto-create	Indica si se crean automáticamente los archivos de la tabla cuando no existen en la ruta especificada.	Boolean	No	false	Valores válidos: false (predeterminado): si los archivos de la tabla de Paimon no existen en la ruta especificada, el trabajo termina con errores. true: si la ruta especificada no existe, Flink crea automáticamente los archivos de la tabla de Paimon.
file.format	Formato de los archivos de datos.	String	No	parquet	Valores válidos: orc parquet avro lance (compatible con Realtime Compute for Apache Flink 11.6 y versiones posteriores) mosaic (compatible con Realtime Compute for Apache Flink 11.9 y versiones posteriores)
bucket	Número de buckets por partición.	Integer	No	1	Paimon distribuye los datos en buckets según el valor de `bucket-key`. Nota Se recomienda que cada bucket contenga menos de 5 GB de datos.
bucket-key	Columnas que se usan como clave del bucket.	String	No	Ninguna	Especifica las columnas que se utilizan para distribuir los datos en buckets. Separe varios nombres de columna con una coma (,). Por ejemplo, `'bucket-key' = 'order_id,cust_id'` distribuye los datos según las columnas `order_id` y `cust_id`. Nota Si no se especifica este parámetro, Paimon distribuye los datos según la clave principal. Si la tabla no tiene clave principal, Paimon distribuye los datos según los valores de todas las columnas.
changelog-producer	Mecanismo de producción del registro de cambios.	String	No	none	Paimon puede generar un registro de cambios completo para cualquier flujo de entrada. Esto significa que cada registro `update_after` tiene un registro `update_before` correspondiente, lo cual simplifica el consumo posterior. Valores válidos: none (predeterminado): no se genera ningún registro de cambios adicional. Los consumidores posteriores aún pueden leer la tabla de Paimon en modo de streaming, pero el registro de cambios está incompleto (contiene solo registros `update_after` sin los registros `update_before` correspondientes). input: escribe el flujo de entrada en archivos de registro de cambios, los cuales funcionan como el registro de cambios completo. full-compaction: genera un registro de cambios completo durante cada compactación total. lookup: genera un registro de cambios completo antes de confirmar cada instantánea. Para obtener más información sobre cómo elegir un productor de registros de cambios, consulte Changelog Production.
full-compaction.delta-commits	Número máximo de confirmaciones entre dos compactaciones totales consecutivas.	Integer	No	Ninguna	Especifica el número máximo de confirmaciones de instantáneas permitidas antes de activar una compactación total.
lookup.cache-max-memory-size	Tamaño de la caché de memoria para una tabla de dimensiones de Paimon.	String	No	256 MB	Este parámetro controla el tamaño de la caché tanto para las búsquedas en tablas de dimensiones como para el productor de registros de cambios `lookup`.
merge-engine	Mecanismo para fusionar registros que comparten la misma clave principal.	String	No	deduplicate	Valores válidos: deduplicate: conserva únicamente el registro más reciente. partial-update: actualiza el registro existente con los valores no nulos del registro más reciente. Las demás columnas permanecen sin cambios. aggregation: realiza una preagregación mediante funciones de agregación específicas. Para consultar un análisis detallado de los motores de fusión, vea Merge Engine.
partial-update.ignore-delete	Indica si se ignoran los mensajes de eliminación (-D).	Boolean	No	false	Valores válidos: true: ignora los mensajes de eliminación. false: procesa los mensajes de eliminación. Debe configurar una estrategia para gestionar las eliminaciones mediante parámetros como sequence.field, con el fin de evitar posibles errores de `IllegalStateException` o `IllegalArgumentException`. Nota En Realtime Compute for Apache Flink 8.0.6 y versiones anteriores, este parámetro surte efecto solo en escenarios de actualización parcial donde `merge-engine = 'partial-update'`. En Realtime Compute for Apache Flink 8.0.7 y versiones posteriores, este parámetro también es compatible con escenarios que no son de actualización parcial y cumple la misma función que el parámetro `ignore-delete`. Se recomienda usar `ignore-delete` en su lugar. Decida si habilita este parámetro según sus requisitos de negocio y si los mensajes de eliminación son esperados. Si dichos mensajes no coinciden con la semántica prevista de su trabajo, suele ser mejor detener la ejecución de inmediato ante un error.
ignore-delete	Indica si se ignoran los mensajes de eliminación (-D).	Boolean	No	false	Los valores válidos son los mismos que los de partial-update.ignore-delete. Nota Este parámetro solo es compatible con Realtime Compute for Apache Flink 8.0.7 y versiones posteriores. Este parámetro cumple la misma función que `partial-update.ignore-delete`. Se recomienda usar `ignore-delete` y evitar configurar ambos parámetros al mismo tiempo.
partition.default-name	Nombre predeterminado de la partición.	String	No	__DEFAULT_PARTITION__	Nombre de partición que se utiliza cuando el valor de una columna de partición es nulo o una cadena vacía.
partition.expiration-check-interval	Frecuencia con la que el sistema comprueba si hay particiones caducadas.	String	No	1h	Para obtener más detalles, consulte How to configure automatic partition expiration.
partition.expiration-time	Tiempo tras el cual caducan las particiones.	String	No	Ninguna	Una partición caduca cuando su antigüedad supera este valor. De forma predeterminada, las particiones nunca caducan. El sistema calcula la antigüedad de una partición a partir de su valor de partición. Para obtener más detalles, consulte How to configure automatic partition expiration.
partition.timestamp-formatter	Cadena de formato para convertir una cadena de tiempo en una marca de tiempo.	String	No	Ninguna	Especifica el formato para extraer la antigüedad de la partición a partir de su valor. Para obtener más detalles, consulte How to configure automatic partition expiration.
partition.timestamp-pattern	Cadena de formato para convertir un valor de partición en una cadena de tiempo.	String	No	Ninguna	Especifica el patrón para extraer una cadena de tiempo del valor de la partición. Para obtener más detalles, consulte How to configure automatic partition expiration.
scan.bounded.watermark	Valor de marca de agua que indica el final de un escaneo. La tabla de origen deja de producir datos cuando su marca de agua supera este valor.	Long	No	Ninguna	N/D
scan.mode	Especifica la posición de consumo para la tabla de origen de Paimon.	String	No	default	Para obtener más detalles, consulte How to set the consumption position for a Paimon source table.
scan.snapshot-id	Especifica la instantánea desde la cual la tabla de origen de Paimon comienza el consumo.	Integer	No	Ninguna	Para obtener más detalles, consulte How to set the consumption position for a Paimon source table.
scan.timestamp-millis	Especifica el momento a partir del cual la tabla de origen de Paimon comienza el consumo.	Integer	No	Ninguna	Para obtener más detalles, consulte How to set the consumption position for a Paimon source table.
snapshot.num-retained.max	Número máximo de instantáneas recientes que se conservan.	Integer	No	2147483647	La expiración de instantáneas se activa si se cumple esta condición o la condición `snapshot.time-retained`, siempre que también se cumpla la condición `snapshot.num-retained.min`.
snapshot.num-retained.min	Número mínimo de instantáneas recientes que se conservan.	Integer	No	10	N/D
snapshot.time-retained	Periodo de retención de las instantáneas.	String	No	1h	La expiración de instantáneas se activa si se cumple esta condición o la condición `snapshot.num-retained.max`, siempre que también se cumpla la condición `snapshot.num-retained.min`.
write-mode	Modo de escritura de la tabla de Paimon.	String	No	change-log	Valores válidos: change-log: la tabla de Paimon admite operaciones de inserción, eliminación y actualización basadas en la clave principal. append-only: la tabla de Paimon acepta únicamente operaciones de inserción y no admite clave principal. Este modo es más eficiente que el modo change-log. Para obtener más información sobre los modos de escritura, consulte Write Mode.
scan.infer-parallelism	Indica si se infiere automáticamente el paralelismo para una tabla de origen de Paimon.	Boolean	No	true	Valores válidos: true: infiere automáticamente el paralelismo de la tabla de origen de Paimon según el número de buckets. false: utiliza el paralelismo predeterminado configurado en Realtime Compute for Apache Flink. Si el modo experto está habilitado, el trabajo usa el paralelismo que usted haya configurado.
scan.parallelism	Paralelismo para la tabla de origen de Paimon.	Integer	No	Ninguna	Nota Este parámetro se ignora si Resource Mode se establece en Mode en la pestaña Configuration > Resources del trabajo.
sink.parallelism	Paralelismo para la tabla de destino de Paimon.	Integer	No	Ninguna	Nota Este parámetro se ignora si Resource Mode se establece en Mode en la pestaña Configuration > Resources del trabajo.
sink.clustering.by-columns	Especifica las columnas de agrupación para escribir en la tabla de destino de Paimon.	String	No	Ninguna	Para Paimon append-only tables (tables without a primary key), este parámetro habilita la escritura agrupada en trabajos por lotes. Este proceso mejora el rendimiento de las consultas al agrupar los datos según las columnas especificadas. Separe varios nombres de columna con una coma (,); por ejemplo, `'col1,col2'`. Para obtener más información sobre la agrupación, consulte la documentación oficial de Apache Paimon.
sink.delete-strategy	Especifica una estrategia de validación para garantizar que el sistema gestione correctamente los mensajes de retracción (-D/-U).	Enum	No	NONE	Valores válidos y comportamiento esperado del operador de destino al gestionar mensajes de retracción: NONE (predeterminado): no aplica ninguna validación. IGNORE_DELETE: el operador de destino debe ignorar los mensajes -U y -D. No se produce ninguna retracción. NON_PK_FIELD_TO_NULL: el operador de destino debe ignorar los mensajes -U. Para los mensajes -D, conserva la clave principal y establece todos los demás campos que no forman parte de ella en null. Se utiliza principalmente para actualizaciones parciales cuando varios destinos escriben en la misma tabla. DELETE_ROW_ON_PK: el operador de destino debe ignorar los mensajes -U, pero eliminar la fila correspondiente a la clave principal cuando recibe mensajes -D. CHANGELOG_STANDARD: el operador de destino debe eliminar la fila correspondiente a la clave principal tanto para los mensajes -U como para los -D. Nota Este parámetro solo es compatible con Realtime Compute for Apache Flink 8.0.8 y versiones posteriores. El comportamiento real del destino ante las retracciones lo determinan otros parámetros, como `ignore-delete` y `merge-engine`. Este parámetro solo valida si el comportamiento real coincide con la estrategia elegida. Si no coinciden, el trabajo termina con errores y muestra un mensaje que sugiere las correcciones necesarias.
blob-as-descriptor	Indica si se generan los bytes del descriptor Blob al leer una columna Blob.	Boolean	No	false	Cuando este parámetro se establece en true, las consultas devuelven los bytes serializados de BlobDescriptor en lugar del contenido real del Blob. Utilice este parámetro junto con la función de URL prefirmada de Blob. No es necesario configurarlo al crear la tabla; puede establecerlo de forma dinámica durante la lectura mediante una sugerencia (hint) de SQL. Este parámetro es compatible con VVR 11.9-preview1 y versiones posteriores. Nota Solo es compatible con VVR 11.9-preview1 y versiones posteriores.

Nota

Para obtener más información sobre las opciones de configuración, consulte la documentación oficial de Apache Paimon.

Parámetros de tablas vectoriales

Importante

Los siguientes parámetros solo son compatibles con VVR 11.8 y versiones posteriores.

Parámetro	Descripción	Tipo de datos	Obligatorio	Valor predeterminado	Observaciones
`ivf.nprobe`	Número de clústeres IVF que se exploran durante la búsqueda.	Integer	No	16	Los valores más altos generalmente mejoran la recuperación, pero aumentan la latencia.
`ivf.refine_factor`	Recupera candidatos IVF equivalentes a top_k × refine_factor y los vuelve a clasificar usando los vectores originales almacenados en la tabla de Paimon.	Float	No	Deshabilitado	Deshabilitado de forma predeterminada para todas las variantes de IVF. Resulta más indicado para índices comprimidos (como ivf-pq e ivf-hnsw-sq) en escenarios donde la recuperación tiene prioridad sobre la latencia.
`hnsw.ef_search`	Amplitud de búsqueda de HNSW durante la consulta.	Integer	No	0	Los valores más altos generalmente mejoran la recuperación, pero aumentan la latencia. El valor 0 indica que se usa el valor predeterminado de la biblioteca nativa.
`diskann.search.list_size`	Tamaño de la lista de búsqueda de Lumina DiskANN.	Integer	No	max(1.5 × top_k, 16)	Los valores más altos generalmente mejoran la recuperación, pero aumentan la latencia.
`diskann.search.beam_width`	Ancho del haz de búsqueda de Lumina DiskANN.	Integer	No	4	—
`search.parallel_number`	Número de búsquedas paralelas de Lumina.	Integer	No	5	—

Detalles de la función

Actualidad y consistencia de los datos

La tabla sink de Paimon utiliza el protocolo de confirmación en dos fases para confirmar los datos durante cada checkpoint del trabajo de Flink. Por lo tanto, la actualidad de los datos depende del intervalo de checkpoint del trabajo de Flink. Cada confirmación genera hasta dos instantáneas.

Cuando dos trabajos de Flink escriben de forma simultánea en la misma tabla de Paimon, logran una consistencia serializable si escriben en distintos buckets. Si escriben en el mismo bucket, solo alcanzan aislamiento de instantánea. Esto significa que los datos de la tabla pueden ser una mezcla de resultados de ambos trabajos, pero no se pierde información.

Motor de fusión

Cuando una tabla sink de Paimon recibe varios registros con la misma clave principal, los fusiona en un solo registro para mantener la unicidad. Este comportamiento se controla al establecer el parámetro merge-engine. A continuación se describen los motores de fusión disponibles.

Motor de fusión	Descripción
Deduplicate	El motor de deduplicación es el predeterminado. Ante varios registros con la misma clave principal, la tabla sink de Paimon conserva únicamente el más reciente y descarta los demás. Nota Si el registro más reciente es un mensaje de eliminación, se descartan todos los registros con esa clave principal.
Partial Update	El motor de actualización parcial permite construir un registro completo mediante actualizaciones incrementales con varios mensajes. Cuando llega un nuevo registro con la misma clave principal, sus valores no nulos sobrescriben los campos correspondientes del registro existente. El motor ignora los campos nulos del nuevo registro y conserva los valores existentes. Por ejemplo, suponga que una tabla sink de Paimon recibe los tres registros siguientes en este orden: <1, 23.0, 10, NULL> <1, NULL, NULL, 'This is a book'> <1, 25.2, NULL, NULL> Si la primera columna es la clave principal, el registro final fusionado es <1, 25.2, 10, 'This is a book'>. Nota Para leer los resultados de una actualización parcial en streaming, debe establecer el parámetro `changelog-producer` en `lookup` o `full-compaction`. El motor de actualización parcial no procesa mensajes de eliminación. Establezca el parámetro `partial-update.ignore-delete` en `true` para ignorarlos.
Aggregation	En algunos casos de uso, solo se necesita el valor agregado de los registros. El motor de agregación combina los registros que comparten la misma clave principal mediante las funciones de agregación que usted especifique. Para cada columna que no sea clave principal, debe especificar una función de agregación con la opción `fields.<field-name>.aggregate-function`. De lo contrario, la columna usará de forma predeterminada la función de agregación `last_non_null_value`. Considere la siguiente definición de tabla de Paimon como ejemplo. `CREATE TABLE MyTable ( product_id BIGINT, price DOUBLE, sales BIGINT, PRIMARY KEY (product_id) NOT ENFORCED ) WITH ( 'merge-engine' = 'aggregation', 'fields.price.aggregate-function' = 'max', 'fields.sales.aggregate-function' = 'sum' );` La columna `price` se agrega con la función `max`, mientras que la columna `sales` se agrega con la función `sum`. Dados dos registros de entrada, <1, 23.0, 15> y <1, 30.2, 20>, el resultado final es <1, 30.2, 35>. Las funciones de agregación compatibles y sus tipos de datos correspondientes son: sum: admite DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT y DOUBLE. min y max: admiten DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP y TIMESTAMP_LTZ. last_value y last_non_null_value: admiten todos los tipos de datos. listagg: admite STRING. bool_and y bool_or: admiten BOOLEAN. Nota Solo la función `sum` admite retracciones y eliminaciones; las demás funciones de agregación no. Si necesita que ciertas columnas ignoren los mensajes de retracción y eliminación, establezca `'fields.${field_name}.ignore-retract'='true'`. Para leer los resultados de una agregación en streaming, debe establecer el parámetro `changelog-producer` en `lookup` o `full-compaction`.

Productor de changelog

Establezca el parámetro changelog-producer para configurar Paimon y generar un changelog completo (donde cada registro update_after tiene su registro update_before correspondiente) para cualquier flujo de entrada. A continuación se describen los productores de changelog disponibles. Para obtener más detalles, consulte la documentación oficial de Apache Paimon.

Productor	Descripción
None	Cuando establece `changelog-producer` en `none` (valor predeterminado), las tablas source de Paimon descendentes solo ven el estado más reciente de los datos para una clave principal dada. Este changelog incompleto dificulta que los consumidores hagan cálculos correctos, ya que no determinan el estado anterior de los datos, sino únicamente si se eliminaron o cuál es su estado más reciente. Por ejemplo, si un consumidor descendente necesita calcular la suma de una columna y solo ve el valor más reciente de 5, no determina cómo actualizar el total. Si el valor anterior era 4, la suma debería aumentar en 1; si era 6, debería disminuir en 1. Los consumidores sensibles a los registros `update_before` no deben usar el productor `none`, aunque los demás productores de changelog introducen sobrecarga de rendimiento. Nota Si su consumidor descendente, como una base de datos, no es sensible a los datos `update_before`, utilice el productor `none`. Configure el productor de changelog según sus requisitos específicos.
Input	Cuando establece `changelog-producer` en `input`, la tabla sink escribe el flujo de entrada de forma duplicada en los archivos de changelog. Utilice este productor solo cuando el flujo de entrada ya sea un changelog completo, como los datos de Change Data Capture (CDC).
Lookup	Cuando establece `changelog-producer` en `lookup`, la tabla sink usa un mecanismo de consulta puntual, similar a la búsqueda en una tabla de dimensiones, para generar un changelog completo de la instantánea actual antes de confirmarla. Este productor genera un changelog completo a partir de cualquier flujo de entrada. En comparación con el productor `full-compaction`, el productor `lookup` ofrece mejor actualidad del changelog, pero consume más recursos en general. Utilice esta opción en casos de uso que requieran alta actualidad de los datos (por ejemplo, a nivel de minutos).
Full Compaction	Cuando establece `changelog-producer` en `full-compaction`, la tabla sink genera un changelog completo durante cada compactación total. Este productor genera un changelog completo a partir de cualquier flujo de entrada. El intervalo de compactación total se especifica con el parámetro `full-compaction.delta-commits`. Frente al productor `lookup`, el productor `full-compaction` presenta mayor latencia, pero aprovecha el proceso de compactación total existente sin añadir cómputo adicional. Esto reduce el consumo general de recursos. Utilice esta opción en casos de uso con requisitos bajos de actualidad de datos (por ejemplo, por hora).

Modo de escritura

Las tablas de Paimon admiten los siguientes modos de escritura.

Modo

Descripción

Change-log

El modo de escritura change-log es el predeterminado para las tablas de Paimon. Admite operaciones de inserción, eliminación y actualización basadas en una clave principal. En este modo también se usan los motores de fusión y los productores de changelog.

Append-only

El modo de escritura append-only solo admite inserciones de datos y no utiliza clave principal. Es más eficiente que el modo change-log y sirve como alternativa a una cola de mensajes en escenarios con requisitos moderados de actualidad de datos (por ejemplo, a nivel de minutos).

Para consultar una descripción detallada del modo de escritura append-only, revise la documentación oficial de Apache Paimon. Al usar este modo, tenga en cuenta lo siguiente:

Establezca el parámetro bucket-key según sus necesidades. De lo contrario, la tabla de Paimon distribuirá los datos en buckets según los valores de todas las columnas, lo cual resulta ineficiente en términos de cómputo.
El modo de escritura append-only garantiza el orden de salida de los datos hasta cierto punto. El orden específico es el siguiente:
1. Para registros de distintas particiones: si se establece el parámetro scan.plan-sort-partition, se emite primero el registro de la partición con el valor menor. De lo contrario, se emite primero el registro de la partición creada antes.
2. Para registros de la misma partición y el mismo bucket, se emite primero el que se escribió antes.
3. Para registros de la misma partición pero de distintos buckets, no se garantiza el orden de salida porque diferentes tareas concurrentes procesan cada bucket.

Destino para la ingesta de datos de Flink CDC

Las tablas de Paimon admiten la sincronización en tiempo real de datos provenientes de tablas individuales o bases de datos completas. Los cambios de esquema ascendentes se sincronizan con las tablas de Paimon en tiempo real. Para obtener más información, consulte la sección Data ingestion.

Poda de lectura de Variant

Lee únicamente los campos de Variant a los que hace referencia la consulta, lo que reduce la sobrecarga de E/S y de memoria.

Cómo habilitarla

Parámetro	Valor predeterminado	Descripción
`variant.read.pushdown.enabled`	false	Establézcalo en `true` para habilitar la poda de lectura de Variant.

Requisitos previos

Los datos de Variant solo se escriben con VVR 11.6 o versiones posteriores. Los datos escritos por versiones anteriores u otros motores no admiten la poda de lectura.
Para que los datos ascendentes surtan efecto, debe habilitar la opción variant.inferShreddingSchema para activar la inference automática de fragmentación (shredding) de Variant, o bien configurar de forma explícita variant.shreddingSchema y las opciones relacionadas para indicar qué miembros de Variant son aptos para la fragmentación. Para obtener más información, consulte Coreoptions.

Escenarios compatibles

La poda de lectura surte efecto cuando SQL accede a campos de Variant mediante una clave de cadena, por ejemplo:

SELECT v['a']            FROM t;   -- Single level
SELECT v['a']['b']       FROM t;   -- Nested
SELECT v['a'], v['b']    FROM t;   -- Multiple fields
SELECT id, v['a']        FROM t;   -- Mixed with regular columns
SELECT v['a'] + 1        FROM t;   -- Referenced in expressions

Escenarios no compatibles

Hacer referencia directa a todo el Variant, como SELECT v FROM t.
Combinar la referencia directa con el acceso a campos, como SELECT v, v['a'] FROM t.
Acceso por índice de arreglo, como v[0].
El esquema de la tabla contiene campos Row anidados. En este caso, la poda de lectura no aplica a ningún campo de Variant de la tabla.

URL prefirmadas de Blob

Paimon genera URL prefirmadas HTTPS GET públicas de corta duración para los datos de columnas Blob almacenados en OSS. Los servicios externos, como los servicios de modelos multimodales, usan estas URL para recuperar el contenido del archivo directamente, sin transferir los bytes dentro de un trabajo de Flink.

Límites

Solo el motor de cómputo de Flink VVR 11.9-preview1 y versiones posteriores admiten esta función.
Solo se admiten tablas de Paimon almacenadas en OSS. El tipo de metastore no importa; se admiten Filesystem, DLF y otros tipos de metastore.
La URL generada es una URL HTTPS pública y debe consumirla un servicio externo con acceso a la red pública. Si el catálogo usa un endpoint interno, se recomienda cambiarlo por un endpoint HTTPS público para que el servicio externo acceda a la URL generada.
Para leer una columna Blob normal, primero debe obtener los bytes de BlobDescriptor mediante la opción blob-as-descriptor y luego pasarlos a la función.

Sintaxis

descriptor_to_presigned_url(source_table, descriptor, validity)
try_descriptor_to_presigned_url(source_table, descriptor, validity)

Parámetros

Parámetro	Tipo	Descripción
`source_table`	Literal STRING	La tabla de Paimon que contiene la columna Blob, con el formato `database.table`. Si el catálogo de la tabla coincide con el catálogo al que pertenece la función, también use `catalog.database.table`. El valor debe ser un literal no vacío. No se admiten campos de tabla, columnas dinámicas ni expresiones CASE.
`descriptor`	BYTES	Los bytes de BlobDescriptor, que resultan de leer la columna Blob tras habilitar `blob-as-descriptor` = 'true'.
`validity`	INTERVAL	El periodo de validez de la URL. El valor debe ser un número positivo de segundos, por ejemplo, `INTERVAL '5' MINUTE`.

Valor devuelto

Tipo	Descripción
STRING	Una URL prefirmada HTTPS GET pública de corta duración.

Ejemplo

-- Use an SQL hint to read the Blob column as a descriptor and generate a presigned URL
SELECT sys.descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

-- Fault-tolerant mode: returns NULL on row-level errors, with other behavior unchanged
SELECT sys.try_descriptor_to_presigned_url(
    'default.image_table',
    image,
    INTERVAL '5' MINUTE)
FROM image_table /*+ OPTIONS('blob-as-descriptor'='true') */;

Nota

descriptor_to_presigned_url lanza una excepción ante un error a nivel de fila. try_descriptor_to_presigned_url devuelve NULL ante un error a nivel de fila.
La URL generada no contiene extensión de archivo, y los bytes a los que apunta son idénticos a los datos de Blob. Los servicios descendentes deben identificar el formato del archivo a partir de los bytes devueltos y no del sufijo de la URL.
Una URL prefirmada equivale a credenciales de acceso temporal. Después de generarla, envíela de inmediato al servicio descendente. No la escriba en registros ni en tablas de resultados, y no la persista.

Tabla de cadena

Una tabla de cadena reconstruye un conjunto de datos completo a partir de una rama de instantánea y los cambios posteriores en una rama delta. Escriba los datos completos periódicos en la rama snapshot y los cambios entre dos cargas completas en la rama delta. Una lectura usa la partición de instantánea más reciente como ancla y fusiona las particiones delta posteriores para reconstruir los datos completos, de modo que ya no es necesario reescribir toda la tabla. Esto resulta útil cuando solo cambia una pequeña fracción de filas entre ejecuciones, como en un volcado de binlog de ODS. Para obtener más información, consulte Chain Table en la documentación de Apache Paimon. Las tablas de cadena solo se admiten en VVR 11.9 y versiones posteriores.

Habilitar una tabla de cadena

Establezca chain-table.enabled en true al crear la tabla, cree las ramas snapshot y delta, y luego configure las opciones de reserva de rama en la tabla principal y en ambas tablas de rama.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `date` STRING,
  PRIMARY KEY (`date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd'
);

-- mycat is the name of the Paimon catalog.
CALL `mycat`.sys.create_branch('default.t', 'snapshot');
CALL `mycat`.sys.create_branch('default.t', 'delta');

ALTER TABLE `default`.`t` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_snapshot` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

ALTER TABLE `default`.`t$branch_delta` SET (
  'scan.fallback-snapshot-branch' = 'snapshot',
  'scan.fallback-delta-branch' = 'delta');

Escribir datos

Escriba los datos completos en t$branch_snapshot. Cada partición de esta rama representa una vista completa en un momento dado. Escriba los datos incrementales en t$branch_delta. Cada partición de esta rama contiene únicamente los cambios de ese periodo.

-- Full write
INSERT OVERWRITE `default`.`t$branch_snapshot` PARTITION (date = '20250810') VALUES ('1', '1', '1');

-- Incremental write
INSERT INTO `default`.`t$branch_delta` PARTITION (date = '20250811') VALUES ('2', '1', '1');

Leer datos

Ruta de lectura	Operación	Resultado
Consulta completa	Consultar una partición de la tabla principal.	Los datos completos de la partición. Se lee directamente si la rama de instantánea la contiene; de lo contrario, se reconstruye a partir de una instantánea ancla y las particiones delta posteriores.
Consulta incremental	Consultar una partición de `t$branch_delta`.	Solo los cambios almacenados en la partición delta seleccionada. No se reconstruye el estado completo.
Consulta híbrida	Combinar una consulta completa y una incremental con `UNION ALL`.	La combinación de resultados completos e incrementales. La misma fila puede aparecer dos veces.
Lectura en streaming	Leer la tabla principal en modo streaming.	Primero se emite el resultado de la fase de carga inicial y luego se consumen de forma continua las nuevas confirmaciones de la rama delta.
Lookup join	Usar la tabla de cadena como tabla de dimensiones.	El estado fusionado de la partición de instantánea más reciente y las particiones delta posteriores de cada grupo, actualizado de forma incremental según `continuous.discovery-interval`.

-- Full query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811';

-- Incremental query
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

-- Hybrid query
SELECT t1, t2, t3 FROM `default`.`t` WHERE date = '20250811'
UNION ALL
SELECT t1, t2, t3 FROM `default`.`t$branch_delta` WHERE date = '20250811';

Lectura en streaming

Un trabajo de lectura en streaming se ejecuta en dos fases:

Fase de carga inicial: de forma predeterminada, el trabajo lee la partición de instantánea más reciente de cada grupo y las particiones delta posteriores. Se excluyen las particiones de instantánea más antiguas. Para fusionar los datos de instantánea y delta por ancla en esta fase y resolver los registros DELETE entre ramas, establezca chain-table.streaming.merge-snapshot en true. La contrapartida es un escaneo más pesado al iniciar el trabajo.
Fase incremental: el trabajo lee de forma continua las nuevas confirmaciones de la rama delta.

SET 'execution.runtime-mode' = 'streaming';

INSERT INTO downstream_sink SELECT * FROM `default`.`t`;

Nota

La fase incremental supervisa únicamente la rama delta. Las escrituras en la rama de instantánea no se detectan hasta que se reinicia el trabajo. Ejecute el procedimiento compact_chain_table de forma regular para fusionar los datos incrementales de una partición en la rama de instantánea. Tras la compactación, solo es necesario fusionar en la fase de carga inicial los cambios delta que llegaron después de dicha compactación.

Varias claves de partición

Si la tabla tiene más de una clave de partición, use chain-table.chain-partition-keys para separar la dimensión de cadena de la dimensión de grupo. El valor debe ser un sufijo contiguo de las claves de partición. Los campos del sufijo forman la cadena ordenada por tiempo, y los campos de partición anteriores se convierten en la dimensión de grupo. Cada grupo de valores de partición mantiene su propia cadena y ancla. Si no se establece esta opción, todas las particiones pertenecen a un único grupo implícito.

CREATE TABLE `default`.`t` (
  `t1` STRING,
  `t2` STRING,
  `t3` STRING,
  `region` STRING,
  `date` STRING,
  PRIMARY KEY (`region`, `date`, `t1`) NOT ENFORCED
) PARTITIONED BY (`region`, `date`) WITH (
  'chain-table.enabled' = 'true',
  'sequence.field' = 't2',
  'bucket-key' = 't1',
  'bucket' = '2',
  'partition.timestamp-pattern' = '$date',
  'partition.timestamp-formatter' = 'yyyyMMdd',
  -- Only date forms the chain. Each region has its own chain.
  'chain-table.chain-partition-keys' = 'date'
);

Limitaciones

Las tablas de cadena solo se admiten para tablas con clave principal. Debe especificar bucket y bucket-key.
El esquema de cada rama debe ser coherente.
Una tabla de cadena requiere sequence.field. Los motores de fusión first-row y aggregation no son compatibles.
La lectura en streaming y el lookup join requieren que la rama delta use el motor de fusión deduplicate (valor predeterminado). La lectura por lotes no se ve afectada.
Los vectores de eliminación solo se admiten en tablas de cadena que usan el motor de fusión deduplicate.
Para changelog-producer solo se admiten none (valor predeterminado) e input. Los modos lookup y full-compaction no son compatibles. Si el valor es none, los registros se normalizan por la clave principal completa que incluye la partición de cadena, y se descartan los registros -D y -U cuyos registros originales residen en otra partición. Establezca el valor en input si los consumidores descendentes deben recibir registros de changelog entre particiones.
La lectura en streaming y el lookup join solo admiten el modo de inicio predeterminado (latest-full). Se produce un error si especifica una posición de inicio como scan.snapshot-id, scan.timestamp-millis, scan.mode establecido en latest o consumer-id. Para hacer una lectura en streaming estándar sin la lógica de tabla de cadena, lea una tabla de rama específica como t$branch_delta.
No se admiten filtros de partición en lecturas en streaming ni en lookup joins de tablas de cadena. Esto incluye columnas de partición en la cláusula WHERE y la opción scan.partitions, ya que un filtro interfiere con la lógica de fusión entre ramas. Use el modo por lotes para leer una partición específica.
La clave de unión de un lookup join no debe contener claves de partición. De lo contrario, se devuelve un error.
partition.timestamp-formatter debe especificar una fecha completa (año-mes-día o año-día del año), usar solo campos de fecha y hora reversibles (y/u, M/L, d, D, H, k, m y s) y tener un paso mínimo de al menos un segundo.

Formato de archivo Mosaic

Mosaic es un formato híbrido columnar y de buckets optimizado para tablas anchas que contienen miles o decenas de miles de columnas. Las columnas se ordenan por nombre y se distribuyen en buckets por rango; se almacenan en formato orientado a columnas dentro de cada bucket y se comprimen de forma independiente con ZSTD. La proyección diferida (projection pushdown) funciona con granularidad de bucket, por lo que una consulta lee únicamente los buckets que contienen las columnas solicitadas. Las columnas con prefijos de nombre similares caen en el mismo bucket, lo que mejora tanto la tasa de compresión como la localidad de las proyecciones. Para obtener más información, consulte Mosaic. El formato Mosaic solo se admite en VVR 11.9 y versiones posteriores.

Habilitar el formato

Establezca file.format en mosaic al crear la tabla. En un trabajo de ingesta de datos, la opción correspondiente es table.properties.file.format.

CREATE TABLE `default`.`wide_table` (
  `id` BIGINT,
  `col_a` STRING,
  `col_b` DOUBLE
) WITH (
  'file.format' = 'mosaic'
);

Opciones de formato

Opción	Valor predeterminado	Descripción
`mosaic.num-buckets`	auto	El número de buckets de columnas para E/S paralela. Si el valor es 0 o no se especifica la opción, el número de buckets se determina automáticamente.
`mosaic.stats-columns`	(vacío)	Los nombres de las columnas para las que se recopilan estadísticas mín/máx destinadas a la evaluación diferida de predicados, separados por comas (,). Si el valor está vacío, no se recopilan estadísticas.
`mosaic.read.prefetch-row-groups`	8	El número de grupos de filas que un lector abre por adelantado respecto al grupo de filas que se está consumiendo. La captación previa superpone la latencia de lectura de rangos con la decodificación. Cada grupo de filas precapturado mantiene su lote decodificado en memoria y usa un flujo de entrada independiente. El valor 0 deshabilita la captación previa.
`mosaic.read.prefetch-max-bytes`	64 mb	El límite superior del tamaño decodificado estimado de los grupos de filas precapturados, calculado a partir de sus recuentos de filas y los tipos de columnas proyectados. Una proyección ancha o un grupo de filas grande reduce el valor efectivo de `mosaic.read.prefetch-row-groups`.

Limitaciones

Mosaic no admite tipos complejos: ARRAY, MAP, MULTISET, ROW, VARIANT, BLOB y VECTOR.

Usar Paimon como tabla de dimensiones

Las tablas de Paimon se utilizan como tablas de dimensiones. Para conocer la sintaxis de JOIN, consulte Dimension table JOIN statements.

De forma predeterminada, el lookup carga todos los datos en cada instancia paralela. Este enfoque solo es adecuado para tablas de dimensiones pequeñas. Para tablas de dimensiones grandes, utilice las soluciones Shuffle Lookup que se describen a continuación.

Tablas de dimensiones particionadas

Si su tabla de dimensiones está particionada y solo necesita datos de la última o las dos últimas particiones, utilice la función de carga dinámica de particiones:

SELECT * FROM T
JOIN DIM /*+ OPTIONS('lookup.dynamic-partition'='max_pt()', 'lookup.dynamic-partition.refresh-interval'='1 h') */
FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col = D.col;

Parámetro	Tipo de datos	Valor predeterminado	Descripción
lookup.dynamic-partition	String	N/A	`max_pt()`: carga solo la partición más reciente. `max_two_pt()`: carga solo las dos particiones más recientes.
lookup.dynamic-partition.refresh-interval	Duration	1 h	El intervalo con el que el sistema comprueba si hay actualizaciones de partición en la tabla de dimensiones.

Tablas de dimensiones grandes: tablas de bucket fijo

Se admite solo en VVR 8.0.8 y versiones posteriores. Para tablas de bucket fijo (bucket > 0), utilice Shuffle Lookup para distribuir los datos por clave de bucket entre las instancias paralelas, de modo que cada instancia cargue únicamente los datos de los buckets asignados:

SELECT /*+ LOOKUP('table'='D', 'shuffle'='true') */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Nota

La clave de unión debe ser la clave de bucket. De forma predeterminada, la clave de bucket es la clave principal.
Solo las tablas de bucket fijo (bucket > 0) admiten esta función.

Tablas de dimensiones grandes: tablas de bucket no fijo

Se admite solo en VVR 8.0.10 y versiones posteriores. Para tablas de bucket dinámico o tablas append, utilice SHUFFLE_HASH o REPLICATED_SHUFFLE_HASH para que cada instancia paralela lea todos los datos, pero conserve únicamente la parte que necesita:

-- Shuffle Hash
SELECT /*+ SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

-- Replicated Shuffle Hash
SELECT /*+ REPLICATED_SHUFFLE_HASH(D) */ T.col1, D.col2
FROM T
JOIN DIM FOR SYSTEM_TIME AS OF T.proc_time AS D
ON T.col1 = D.col1;

Para obtener más información sobre SHUFFLE_HASH y REPLICATED_SHUFFLE_HASH, consulte Dimension table JOIN statements.

Ingesta de datos

Utilice el conector de Paimon como sink en trabajos YAML de ingesta de datos.

Sintaxis

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

Parámetros

Parámetro	Descripción	Obligatorio	Tipo	Predeterminado	Notas
type	El tipo de conector.	Sí	STRING	Ninguna	El valor debe ser `paimon`.
name	El nombre del sink.	No	STRING	Ninguna
catalog.properties.metastore	El tipo de catálogo de Paimon.	No	STRING	filesystem	Valores válidos: filesystem (predeterminado) rest (solo es compatible con Data Lake Formation (DLF), no con DLF-Legacy)
catalog.properties.*	Parámetros para crear un catálogo de Paimon.	No	STRING	Ninguna	Para más información, consulte Manage Paimon Catalog.
table.properties.*	Parámetros para crear una tabla de Paimon.	No	STRING	Ninguna	Para más información, consulte Opciones de tabla de Paimon.
table.properties.file.format	El formato de almacenamiento de los archivos de datos en la tabla.	String	No	parquet	Valores válidos: orc parquet avro lance (VVR 11.6 y posteriores) mosaic (VVR 11.9 y posteriores)
catalog.properties.warehouse	El directorio raíz para el almacenamiento de archivos.	No	STRING	Ninguna	Este parámetro solo aplica cuando `catalog.properties.metastore` se establece en `filesystem`.
commit.user-prefix	El prefijo del nombre de usuario para confirmar los archivos de datos.	No	STRING	Ninguna	Nota Se recomienda establecer nombres de usuario distintos para cada job. Esto facilita identificar qué job provoca un conflicto de confirmación.
partition.key	Las claves de partición de una tabla particionada.	No	STRING	Ninguna	Las tablas se separan con `;`, los campos con `,` y las tablas de los campos con `:`. Por ejemplo: `testdb.table1:id1,id2;testdb.table2:name`.
sink.cross-partition-upsert.tables	Enumera las tablas que requieren un upsert entre particiones, donde la clave primaria no incluye todas las claves de partición.	No	STRING	Ninguna	Aplica a tablas con actualizaciones entre particiones. Formato: use un punto y coma `;` para separar los nombres de las tablas. Recomendación de rendimiento: esta operación consume muchos recursos. Cree jobs independientes para estas tablas. Importante Debe enumerar todas las tablas que cumplan con los criterios. Si omite un nombre de tabla, se duplicarán los datos.
sink.commit.parallelism	Especifica el paralelismo del operador Commit.	No	INTEGER	Ninguna	Si el operador Commit representa un cuello de botella, utilice este parámetro para aumentar su paralelismo y mejorar el rendimiento. Este parámetro solo es compatible con Realtime Compute for Apache Flink 11.6 y versiones posteriores. Nota Establecer este parámetro modifica el paralelismo del operador. Al reiniciar un job con estado, debe especificar `AllowNonRestoredState` para que el job ignore el estado parcial del operador.
sink.existing-table.schema-check.enabled	Indica si se comprueba la compatibilidad del esquema de la tabla existente y si se adapta automáticamente al escribir datos en una tabla creada previamente.	Boolean	No	true	A continuación se describen los valores válidos y el comportamiento al escribir datos en una tabla creada previamente: true (predeterminado): comprueba la compatibilidad de los tipos de columna y de las columnas NOT NULL faltantes o redundantes en la tabla existente. Además, agrega automáticamente las columnas anulables presentes en la tabla de origen pero ausentes en la de destino. Si un caso no se adapta automáticamente, el job falla y devuelve las instrucciones SQL que puede ejecutar para corregir el problema. false: omite la comprobación anterior y la adición automática de columnas. Los datos se escriben según el esquema existente de la tabla de destino. Solo se omite la adición automática de columnas durante la ejecución de la instrucción CREATE TABLE inicial. Los cambios de esquema posteriores se aplican automáticamente. Nota Este parámetro solo es compatible con Realtime Compute for Apache Flink VVR 11.8 y versiones posteriores.

Reutilizar un catálogo existente

A partir de Realtime Compute for Apache Flink 11.5, puede hacer referencia directa a un catálogo de Paimon integrado desde la página Data Management en un job de ingesta de datos de Flink CDC. Esto reduce la configuración manual.

sink:
  type: paimon
  using.built-in-catalog: paimon_dlf_catalog
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com

Los jobs de ingesta de datos reutilizan automáticamente todos los parámetros de un catálogo de Paimon. Esto equivale a configurar manualmente los parámetros con el prefijo catalog.properties. en un job YAML.

Para anular un parámetro reutilizado automáticamente, establézcalo de forma explícita en el job YAML. La configuración explícita en YAML tiene mayor prioridad. Por ejemplo, en la muestra anterior, el parámetro fs.oss.endpoint usa el valor del job YAML y anula el de paimon_dlf_catalog.

Cambios de esquema

Un job de pipeline YAML de CDC utiliza distintas estrategias para gestionar los cambios de esquema. Para configurar la estrategia, establezca el parámetro de nivel de pipeline schema.change.behavior como se describe en Schema evolution configurations. El parámetro schema.change.behavior admite los valores IGNORE, LENIENT, TRY_EVOLVE, EVOLVE o EXCEPTION. En las siguientes secciones se describe cómo los modos LENIENT y EVOLVE, que implican modificaciones de esquema, gestionan los distintos eventos de cambio.

LENIENT (predeterminado)

En el modo LENIENT se admiten los siguientes cambios de esquema:

Agregar una columna anulable: la columna se anexa automáticamente al esquema de la tabla de resultados y sus datos se sincronizan de forma automática.
Eliminar una columna anulable: la columna no se elimina de la tabla de resultados. En su lugar, los datos de la columna se reemplazan automáticamente por valores NULL.
Agregar una columna NOT NULL: la columna se anexa automáticamente al esquema de la tabla de resultados y sus datos se sincronizan de forma automática. De manera predeterminada, la nueva columna es anulable y los datos generados antes de agregarla se establecen en NULL.
Cambiar el nombre de una columna: la operación se trata como la adición de una columna y la eliminación de otra. La columna renombrada se anexa al final de la tabla de resultados y los datos de la columna original se reemplazan automáticamente por valores NULL. Por ejemplo, si se cambia el nombre de col_a a col_b, col_b se anexa al final de la tabla de resultados y los datos de col_a se sustituyen por valores NULL.
Modificar el tipo de una columna: cambia el tipo de datos de la columna correspondiente en la tabla de resultados.
No se admiten los siguientes cambios de esquema:
- Modificaciones en los tipos de clave primaria.
- Cambios de NULLABLE a NOT NULL.

EVOLVE

En el modo EVOLVE se admiten los siguientes cambios de esquema:

Agregar una columna anulable: compatible.
Eliminar una columna anulable: no compatible.
Agregar una columna NOT NULL: se agrega una columna anulable a la tabla de resultados.
Cambiar el nombre de una columna: compatible. Se renombra la columna original en la tabla de resultados.
Modificar el tipo de una columna: cambia el tipo de datos de la columna correspondiente en la tabla de resultados.
No se admiten los siguientes cambios de esquema:
- Modificaciones en los tipos de clave primaria.
- Cambios de NULLABLE a NOT NULL.

Para ver un ejemplo sobre cómo habilitar el modo EVOLVE, consulte Enabling the EVOLVE mode.

Nota

Si la tabla de Paimon de destino ya existe, el job escribe los datos en el esquema de dicha tabla sin intentar crearla de nuevo. Cuando el esquema existente difiere del de la tabla de origen, las columnas anulables presentes en el origen pero ausentes en el destino se agregan automáticamente a la tabla existente. Si un caso no se adapta de forma automática, como un tipo de columna incompatible o una columna NOT NULL faltante o redundante, la comprobación falla y se reporta un error. Puede ejecutar las instrucciones SQL recomendadas en el mensaje de error para modificar el esquema de la tabla. También puede establecer sink.existing-table.schema-check.enabled en false dentro del sink para omitir la comprobación y la adición automática de columnas. De este modo, los datos se escriben según el esquema de la tabla de destino.

Ejemplos

A continuación se muestran configuraciones para escenarios típicos.

Escribir en un Rest Catalog

El siguiente ejemplo muestra cómo escribir datos en Data Lake Formation (DLF) cuando el catálogo de Paimon es de tipo rest:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Para obtener información sobre los parámetros con el prefijo catalog.properties, consulte Flink CDC Catalog configuration parameters.

Escribir en un FileSystem Catalog

Configuración de ejemplo para escribir en Object Storage Service (OSS) con un catálogo de Paimon de tipo filesystem:

source:
  type: mysql
  name: MySQL Source
  hostname: ${secret_values.mysql.hostname}
  port: ${mysql.port}
  username: ${secret_values.mysql.username}
  password: ${secret_values.mysql.password}
  tables: ${mysql.source.table}
  server-id: 8601-8604
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: filesystem
  catalog.properties.warehouse: oss://default/test
  catalog.properties.fs.oss.endpoint: oss-cn-beijing-internal.aliyuncs.com
  catalog.properties.fs.oss.accessKeyId: xxxxxxxx
  catalog.properties.fs.oss.accessKeySecret: xxxxxxxx
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

Para obtener información sobre los parámetros con el prefijo catalog.properties, consulte Create a Paimon Filesystem Catalog.

Escribir en una tabla particionada

Convierte el campo create_time de tipo TIMESTAMP en la tabla de origen a un campo DATE, el cual se utiliza como clave de partición de la tabla de Paimon.

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
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job 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: 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 Paimon Pipeline

Sincronizar una sola tabla

Se recomienda configurar jobs de sincronización de tabla única para aquellas tablas con requisitos estrictos de rendimiento en tiempo real y estabilidad. La sincronización de una sola tabla evita la complejidad de sincronizar varias tablas. A continuación se muestra la configuración:

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
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

Sincronizar una base de datos completa

Sincronizar una base de datos completa reduce la cantidad de jobs que debe configurar y ejecutar, lo que disminuye los costos de operación y mantenimiento. A continuación se muestra la configuración:

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

Sincronizar una base de datos completa y reemplazar nombres de bases de datos y tablas

Para reemplazar varios nombres de bases de datos y tablas a la vez, agregue la sección route a un job de sincronización de base de datos completa. A continuación se muestra la configuración:

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # Synchronize all tables in the MySQL test_db database to the Paimon test_db2 database and retain the table names.
  - source-table: test_db.\.*
    sink-table: test_db2.<>
    replace-symbol: <>

pipeline:
  name: MySQL to Paimon Pipeline

Combinar bases de datos y tablas fragmentadas

Para escribir datos de varias tablas fragmentadas en una sola tabla de destino, utilice la siguiente plantilla:

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.user\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  
route:
  # All sharded tables in the MySQL test_db database are merged into a single Paimon test_db.user table.
  - source-table: test_db.user\.*
    sink-table: test_db.user

pipeline:
  name: MySQL to Paimon Pipeline

Agregar tablas existentes al reiniciar el job

Para agregar tablas existentes a la sincronización, establezca scan.newly-added-table.enabled = true y reinicie el job.

Importante

Si el job primero establece scan.binlog.newly-added-table.enabled en true para capturar tablas nuevas, no configure después scan.newly-added-table.enabled en true ni reinicie el job para capturar tablas existentes. De lo contrario, los datos se entregarán más de una vez.

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 job restart, check the new tables captured by the tables parameter and take a snapshot.
  # Note: Must be used together with scan.startup.mode: initial.
  scan.newly-added-table.enabled: true
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

Excluir tablas específicas en la sincronización de una base de datos completa

Para excluir tablas específicas de un job de sincronización de base de datos completa, como tablas de prueba o aquellas que provocan failovers repetidos del job, utilice la siguiente configuración:

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  # Tables that match this regular expression are not synchronized.
  tables.exclude: test_db.table1
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

pipeline:
  name: MySQL to Paimon Pipeline

Sincronizar una base de datos completa en tablas de Paimon con formato de archivo Lance

Lance es un formato de almacenamiento de datos diseñado para datos vectoriales y multimodales. Establezca el parámetro de creación de tabla file.format para escribir datos en tablas de Paimon que usen el formato de archivo Lance. A continuación se muestra la configuración:

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: flink
  password: ${secret_values.password}
  tables: test_db.\.*
  server-id: 5401-5499
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true
  # Specify the file format of output files as lance 
  table.properties.file.format: lance
pipeline:
  name: MySQL to Paimon Pipeline

Habilitar el modo EVOLVE

El modo EVOLVE mantiene el esquema de la tabla de destino estrictamente consistente con el de la tabla de origen, incluidas operaciones como eliminar campos y eliminar tablas. En este modo, si el destino no aplica todos los eventos de cambio de esquema, el job podría sufrir un failover y no recuperarse sin intervención. A continuación se muestra la configuración del job:

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
  #(Optional) Synchronize data from newly added tables during the incremental phase.
  scan.binlog.newly-added-table.enabled: true
  #(Optional) Synchronize table and field comments.
  include-comments.enabled: true
  #(Optional) Prioritize the distribution of unbounded shards to prevent potential Out Of Memory (OOM) issues on TaskManagers.
  scan.incremental.snapshot.unbounded-chunk-first.enabled: true
  #(Optional) Enable parsing and filtering to accelerate data reads.
  scan.only.deserialize.captured.tables.changelog.enabled: true

sink:
  type: paimon
  name: Paimon Sink
  using.built-in-catalog: paimon_dlf_catalog
  #(Optional) Specify a commit user. We recommend that you use a different commit user for each job to avoid conflicts.
  commit.user: your_job_name
  #(Optional) Enable deletion vectors to improve read performance.
  table.properties.deletion-vectors.enabled: true

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

Contexto

Funciones principales

Límites y recomendaciones

SQL

Sintaxis

Parámetros WITH

Parámetros de tablas vectoriales

Detalles de la función

Actualidad y consistencia de los datos

Motor de fusión

Productor de changelog

Modo de escritura

Destino para la ingesta de datos de Flink CDC

Poda de lectura de Variant

URL prefirmadas de Blob

Tabla de cadena

Formato de archivo Mosaic

Usar Paimon como tabla de dimensiones

Tablas de dimensiones particionadas

Tablas de dimensiones grandes: tablas de bucket fijo

Tablas de dimensiones grandes: tablas de bucket no fijo

Ingesta de datos

Sintaxis

Parámetros

Reutilizar un catálogo existente

Cambios de esquema

LENIENT (predeterminado)

EVOLVE

Ejemplos

Escribir en un Rest Catalog

Escribir en un FileSystem Catalog

Escribir en una tabla particionada

Sincronizar una sola tabla

Sincronizar una base de datos completa

Sincronizar una base de datos completa y reemplazar nombres de bases de datos y tablas

Combinar bases de datos y tablas fragmentadas

Agregar tablas existentes al reiniciar el job

Excluir tablas específicas en la sincronización de una base de datos completa

Sincronizar una base de datos completa en tablas de Paimon con formato de archivo Lance

Habilitar el modo EVOLVE

Preguntas frecuentes