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

El conector de OceanBase integra Realtime Compute for Apache Flink con OceanBase, una base de datos nativa distribuida de procesamiento transaccional y analítico híbrido (HTAP). Utilícelo para leer flujos de captura de datos de cambio (CDC), unir tablas de dimensiones y escribir resultados nuevamente en OceanBase.

El conector de OceanBase se encuentra en versión preliminar pública.

Contexto

OceanBase es un sistema de gestión de bases de datos nativo y distribuido para el procesamiento transaccional y analítico híbrido (HTAP). Para obtener más información, consulte el sitio web de OceanBase. Para reducir el costo de refactorización de sus sistemas empresariales durante la migración desde una base de datos MySQL u Oracle, OceanBase admite dos modos de compatibilidad: modo Oracle y modo MySQL. Los tipos de datos, las funciones SQL y las vistas internas en estos modos son consistentes con los de MySQL u Oracle.

Funciones compatibles

Categoría	Detalles
Tipos de tabla	Tablas de origen, de dimensión y de destino
Modos de ejecución	Modo streaming y modo por lotes
Formato de datos	No aplicable
Métricas de monitoreo específicas	Ninguna
API	SQL
Actualizaciones y eliminaciones en tablas de destino	Sí

Requisitos previos

Antes de empezar, asegúrese de que:

La base de datos y las tablas de destino existen en OceanBase
Se ha configurado una lista de permitidos de direcciones IP; consulte Configure a whitelist group
(Para tablas de origen CDC) La función OceanBase Binlog service está habilitada; consulte Binlog-related operations
(Para tablas de destino de importación directa) El puerto de importación directa está habilitado; consulte Bypass import

Limitaciones

Requiere Ververica Runtime (VVR) 8.0.1 o posterior.

Garantías semánticas:

Tablas de origen CDC: semántica exactly-once. Los datos no se pierden ni se duplican durante la transición de la lectura de datos históricos completos a la lectura de Binlog, incluso después de un fallo.
Tablas de destino: semántica at-least-once. Si la tabla de destino tiene una clave primaria, la idempotencia garantiza la corrección de los datos.

Cambio en la arquitectura CDC de VVR 11.4.0

A partir de VVR 11.4.0, el conector CDC de OceanBase se actualizó:

El conector CDC original basado en el servicio LogProxy de OceanBase está obsoleto y se ha eliminado.
La captura incremental de registros ahora requiere el servicio OceanBase Binlog service. El conector CDC de OceanBase ofrece mejor compatibilidad de protocolos y estabilidad de conexión con el servicio Binlog en comparación con la conexión directa del conector CDC estándar de MySQL. No se recomienda conectar el conector CDC estándar de MySQL al servicio Binlog de OceanBase para el seguimiento de cambios.
El seguimiento incremental de cambios en el modo de compatibilidad con Oracle ya no es compatible. Para CDC en modo Oracle, póngase en contacto con el soporte técnico empresarial de OceanBase.

Sintaxis

CREATE TABLE oceanbase_source (
   order_id     INT,
   order_date   TIMESTAMP(0),
   customer_name STRING,
   price        DECIMAL(10, 5),
   product_id   INT,
   order_status BOOLEAN,
   PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
  'connector'  = 'oceanbase',
  'url'        = '<your-jdbc-url>',
  'tableName'  = '<your-table-name>',
  'userName'   = '<your-username>',
  'password'   = '<your-password>'
);

Comportamiento de escritura en el destino:

Para cada registro entrante, el conector construye una sentencia SQL basada en el esquema de la tabla de destino:

Sin clave primaria: INSERT INTO
Con clave primaria: UPSERT (según el modo de compatibilidad de la base de datos)

Parámetros WITH

Parámetros generales

Estos parámetros aplican a todos los tipos de tabla.

Parámetro	Descripción	Obligatorio	Tipo	Valor predeterminado
`connector`	Establezca este valor en `oceanbase`.	Sí	STRING	—
`password`	La contraseña de la base de datos.	Sí	STRING	—

Parámetros de tabla de origen

Importante

A partir de VVR 11.4.0, el conector CDC de OceanBase utiliza el servicio Binlog de OceanBase para la captura incremental de registros. El conector basado en LogProxy se ha eliminado. El modo CDC de compatibilidad con Oracle ya no es compatible desde VVR 11.4.0.

Parámetro	Descripción	Obligatorio	Tipo	Valor predeterminado	Notas
`hostname`	La dirección IP o nombre de host de la base de datos OceanBase. Utilice una dirección de nube privada virtual (VPC) siempre que sea posible.	Sí	STRING	—	Si OceanBase y Realtime Compute for Apache Flink están en diferentes VPC, configure la conectividad entre VPC o utilice un endpoint público. Consulte Workspace management y Internet access for Flink clusters.
`username`	El nombre de usuario de la base de datos OceanBase.	Sí	STRING	—	—
`database-name`	El nombre de la base de datos OceanBase. Admite expresiones regulares para leer desde varias bases de datos. Evite los anclajes `^` y `$`.	Sí	STRING	—	El conector concatena `database-name` y `table-name` con `\\.` (VVR 8.0.1+) o `.` (versiones anteriores) para formar una expresión regular de ruta completa. Por ejemplo, `db_.` + `tb_.+` se convierte en `db_.\\.tb_.+`.
`table-name`	El nombre de la tabla OceanBase. Admite expresiones regulares para leer desde varias tablas. Evite los anclajes `^` y `$`.	Sí	STRING	—	Consulte la nota sobre `database-name` arriba.
`port`	El puerto de la base de datos OceanBase.	No	INTEGER	`3306`	—
`server-id`	Un ID numérico para el cliente de la base de datos. Debe ser único globalmente. Admite un rango, como `5400-5408`, para asignar diferentes ID a lectores concurrentes.	No	STRING	Valor aleatorio entre 5400 y 6400	Utilice un ID diferente para cada trabajo que se conecte a la misma base de datos. Consulte Server ID usage.
`scan.incremental.snapshot.chunk.size`	El número de filas por fragmento durante la lectura de instantáneas incrementales. Los datos de cada fragmento se almacenan en búfer en memoria antes de leerse por completo. Fragmentos más pequeños mejoran la granularidad de recuperación ante fallos, pero pueden causar errores de falta de memoria (OOM) y reducir el throughput.	No	INTEGER	`8096`	Equilibre el tamaño del fragmento con los requisitos de memoria y throughput.
`scan.snapshot.fetch.size`	El número máximo de registros obtenidos por extracción durante las lecturas completas de tabla.	No	INTEGER	`1024`	—
`scan.startup.mode`	El modo de inicio para el consumo de datos.	No	STRING	`initial`	Valores válidos: `initial` (escanea el historial completo y luego lee Binlog), `latest-offset` (solo la cola de Binlog), `earliest-offset` (el Binlog disponible más antiguo), `specific-offset` (configurado mediante los parámetros `scan.startup.specific-offset.` ), `timestamp` (configurado mediante `scan.startup.timestamp-millis` ). Importante* Para los modos `earliest-offset`, `specific-offset` y `timestamp`, el esquema de la tabla no debe cambiar entre la posición de Binlog especificada y el inicio del trabajo.
`scan.startup.specific-offset.file`	El nombre del archivo Binlog para el desplazamiento inicial. Ejemplo: `mysql-bin.000003`.	No	STRING	—	Requiere `scan.startup.mode=specific-offset`.
`scan.startup.specific-offset.pos`	El desplazamiento en bytes dentro del archivo Binlog especificado.	No	INTEGER	—	Requiere `scan.startup.mode=specific-offset`.
`scan.startup.specific-offset.gtid-set`	El conjunto GTID para el desplazamiento inicial. Ejemplo: `24DA167-0C0C-11E8-8442-00059A3C7B00:1-19`.	No	STRING	—	Requiere `scan.startup.mode=specific-offset`.
`scan.startup.timestamp-millis`	La marca de tiempo de inicio en milisegundos. OceanBase CDC lee el evento inicial de cada archivo Binlog para localizar el archivo que coincide con esta marca de tiempo. El archivo Binlog no debe haberse purgado.	No	LONG	—	Requiere `scan.startup.mode=timestamp`.
`server-time-zone`	La zona horaria de sesión utilizada por la base de datos. Controla cómo se convierten los tipos `TIMESTAMP` a `STRING`. Consulte tipos temporales de Debezium. Ejemplo: `Asia/Shanghai`.	No	STRING	Zona horaria de ejecución del trabajo de Flink	—
`debezium.min.row.count.to.stream.results`	El umbral de recuento de filas por encima del cual el conector cambia de lectura completa (toda la tabla en memoria) a lectura por lotes (transmisión de filas en lotes). La lectura completa es más rápida; la lectura por lotes evita errores OOM en tablas grandes.	No	INTEGER	`1000`	—
`connect.timeout`	El tiempo máximo de espera antes de reintentar una conexión que ha expirado.	No	DURATION	`30s`	—
`connect.max-retries`	El número máximo de reintentos de conexión tras un fallo.	No	INTEGER	`3`	—
`connection.pool.size`	El tamaño del pool de conexiones de la base de datos. Reutilizar conexiones reduce el número total de conexiones abiertas.	No	INTEGER	`20`	—
`jdbc.properties.*`	Parámetros de conexión personalizados para la URL JDBC. Ejemplo: `'jdbc.properties.useSSL' = 'false'`. Consulte propiedades de configuración de MySQL.	No	STRING	—	—
`debezium.*`	Parámetros personalizados de Debezium para la lectura de Binlog. Ejemplo: `'debezium.event.deserialization.failure.handling.mode' = 'ignore'`.	No	STRING	—	—
`heartbeat.interval`	El intervalo en el que el origen emite eventos de latido para avanzar el desplazamiento de Binlog. Evita que el desplazamiento de Binlog expire en tablas que se actualizan con poca frecuencia. Un desplazamiento expirado provoca el fallo del trabajo y requiere un reinicio sin estado.	No	DURATION	`30s`	—
`scan.incremental.snapshot.chunk.key-column`	La columna utilizada como clave de fragmento para dividir los datos durante la fase de instantánea.	Condicional	STRING	—	Obligatorio para tablas sin clave primaria (debe ser `NOT NULL`). Opcional para tablas con clave primaria (seleccione una columna de la clave primaria).
`scan.incremental.close-idle-reader.enabled`	Especifica si se deben cerrar los lectores inactivos después de que finalice la fase de instantánea.	No	BOOLEAN	`false`	VVR 8.0.1+. También requiere `execution.checkpointing.checkpoints-after-tasks-finish.enabled=true`.
`scan.read-changelog-as-append-only.enabled`	Especifica si se debe convertir el flujo de registro de cambios en un flujo de solo anexión. Cuando es `true`, todos los tipos de mensajes (`INSERT`, `DELETE`, `UPDATE_BEFORE`, `UPDATE_AFTER`) se convierten a `INSERT`. Habilítelo solo en escenarios especiales, como preservar mensajes de eliminación de una tabla ascendente.	No	BOOLEAN	`false`	VVR 8.0.8+.
`scan.only.deserialize.captured.tables.changelog.enabled`	Especifica si se deben deserializar eventos de cambio solo para las tablas capturadas durante la fase incremental. Establecerlo en `true` acelera la lectura de Binlog.	No	BOOLEAN	`false` (VVR 8.x), `true` (VVR 11.1+)	VVR 8.0.7+. En VVR 8.0.8 y anteriores, utilice el nombre de parámetro `debezium.scan.only.deserialize.captured.tables.changelog.enable`.
`scan.parse.online.schema.changes.enabled`	Especifica si se deben analizar eventos DDL para cambios sin bloqueo de ApsaraDB RDS durante la fase incremental. Función experimental. Tome una instantánea del trabajo de Flink antes de realizar cambios de esquema en línea sin bloqueo.	No	BOOLEAN	`false`	VVR 11.1+.
`scan.incremental.snapshot.backfill.skip`	Especifica si se debe omitir el relleno durante la fase de instantánea. El relleno se aplica únicamente durante la consulta de instantánea de un solo fragmento y no cubre toda la fase de lectura completa. Cuando se omite el relleno, la consulta de instantánea de cada fragmento lee los datos más recientes de la tabla en ese instante; las actualizaciones que ocurren en un fragmento después de haber sido leído no se fusionan durante la fase de lectura completa y se leen desde Binlog después de entrar en la fase incremental. Por ejemplo, una actualización en chunk5 que ocurre mientras se toma la instantánea de chunk5 se refleja directamente en la instantánea de chunk5; si chunk5 se actualiza después de que el lector haya avanzado a chunk80, la actualización se aplica posteriormente desde Binlog durante la fase incremental. Importante: cuando está habilitado, los cambios que ocurren durante o después del escaneo de un fragmento todavía se entregan desde Binlog en la fase incremental y pueden duplicarse; solo se garantiza la semántica at-least-once. Habilite esto únicamente cuando el destino descendente admita escrituras idempotentes por clave primaria.	No	BOOLEAN	`false`	VVR 11.1+.
`scan.incremental.snapshot.unbounded-chunk-first.enabled`	Especifica si se debe distribuir primero el fragmento ilimitado durante la fase de instantánea. Reduce el riesgo de OOM cuando un TaskManager procesa el último fragmento. Función experimental. Agregue este parámetro antes de que el trabajo se inicie por primera vez.	No	BOOLEAN	`false`	VVR 11.1+.

Parámetros de tabla de dimensión

Parámetro	Descripción	Obligatorio	Tipo	Valor predeterminado	Notas
`url`	La URL JDBC. Debe incluir el nombre de la base de datos MySQL o el nombre del servicio Oracle.	Sí	STRING	—	—
`userName`	El nombre de usuario de la base de datos.	Sí	STRING	—	—
`cache`	La directiva de caché para búsquedas en tablas de dimensión.	No	STRING	`ALL`	`ALL`: Carga todos los datos antes de iniciar el trabajo; recarga tras la expiración. Adecuado para tablas pequeñas con muchas búsquedas fallidas. Aumente la memoria del nodo de unión a al menos el doble del tamaño de la tabla para admitir la carga asíncrona. `LRU`: Almacena en caché un subconjunto de filas; requiere `cacheSize`. `None`: Sin almacenamiento en caché.
`cacheSize`	El número máximo de entradas en caché.	No	INTEGER	`100000`	Obligatorio cuando `cache=LRU`. Se ignora cuando `cache=ALL`.
`cacheTTLMs`	El tiempo de expiración de la caché en milisegundos. El comportamiento depende de la configuración de `cache`: para `LRU`, las entradas expiran tras esta duración (sin expiración por defecto); para `ALL`, la caché completa se recarga tras esta duración (sin recarga por defecto); para `None`, este parámetro no tiene efecto.	No	LONG	`Long.MAX_VALUE`	—
`maxRetryTimeout`	La duración máxima de reintento.	No	DURATION	`60s`	—

Parámetros de tabla de destino (JDBC)

Parámetro	Descripción	Obligatorio	Tipo	Valor predeterminado	Notas
`url`	La URL JDBC. Debe incluir el nombre de la base de datos MySQL o el nombre del servicio Oracle.	Sí	STRING	—	—
`userName`	El nombre de usuario de la base de datos.	Sí	STRING	—	—
`tableName`	El nombre de la tabla de destino.	Sí	STRING	—	—
`sink.mode`	El modo de escritura. Establezca en `jdbc` para escrituras estándar; establezca en `direct-load` para importación directa.	Sí	STRING	`jdbc`	—
`compatibleMode`	El modo de compatibilidad de OceanBase. Valores válidos: `mysql`, `oracle`.	No	STRING	`mysql`	Parámetro específico de OceanBase.
`maxRetryTimes`	El número máximo de reintentos de escritura.	No	INTEGER	`3`	—
`poolInitialSize`	El tamaño inicial del pool de conexiones.	No	INTEGER	`1`	—
`poolMaxActive`	El número máximo de conexiones activas en el pool.	No	INTEGER	`8`	—
`poolMaxWait`	El tiempo máximo de espera (ms) para obtener una conexión del pool.	No	INTEGER	`2000`	—
`poolMinIdle`	El número mínimo de conexiones inactivas en el pool.	No	INTEGER	`1`	—
`connectionProperties`	Propiedades de conexión JDBC en formato `k1=v1;k2=v2`.	No	STRING	—	—
`ignoreDelete`	Especifica si se deben ignorar las operaciones de eliminación.	No	BOOLEAN	`false`	—
`excludeUpdateColumns`	Columnas a excluir de las actualizaciones, separadas por comas (por ejemplo, `column1,column2`). Las columnas de clave primaria siempre se excluyen independientemente de esta configuración.	No	STRING	—	—
`partitionKey`	La clave de partición. Cuando se establece, los datos se agrupan por esta clave antes de aplicar la regla `modRule`.	No	STRING	—	—
`modRule`	La regla de agrupación en formato `column_name mod number` (por ejemplo, `user_id mod 8`). La columna debe ser numérica. Los datos se particionan primero por `partitionKey` y luego se agrupan dentro de cada partición según esta regla.	No	STRING	—	—
`bufferSize`	El tamaño del búfer de datos (número de registros).	No	INTEGER	`1000`	—
`flushIntervalMs`	El intervalo de vaciado del búfer (ms). Si el búfer no cumple la condición de salida dentro de este intervalo, todos los datos almacenados se vacían automáticamente.	No	LONG	`1000`	—
`retryIntervalMs`	El intervalo de reintento (ms).	No	INTEGER	`5000`	—

Parámetros de tabla de destino (importación directa)

La importación directa es un método de escritura de alto throughput diseñado para la carga masiva de datos en OceanBase. Disponible en VVR 11.5 y versiones posteriores.

Antes de utilizar la importación directa, verifique las siguientes restricciones:

Solo flujos acotados: El origen de datos debe ser un flujo acotado. Utilice el modo por lotes de Flink para obtener el mejor rendimiento.
Bloqueo de tabla durante la importación: La tabla de destino permanece bloqueada durante toda la importación. Las escrituras DML y los cambios DDL se bloquean; las consultas de lectura no se ven afectadas.
No apto para escrituras en tiempo real: Para escrituras en streaming o en tiempo real, utilice el destino JDBC.

Parámetro	Descripción	Obligatorio	Tipo	Valor predeterminado	Notas
`sink.mode`	Establezca en `direct-load` para usar la importación directa.	No	STRING	`jdbc`	—
`host`	La dirección IP o nombre de host de la base de datos OceanBase.	Sí	STRING	—	—
`port`	El puerto RPC de la base de datos OceanBase.	No	INTEGER	`2882`	—
`username`	El nombre de usuario de la base de datos.	Sí	STRING	—	—
`tenant-name`	El nombre del inquilino de OceanBase.	Sí	STRING	—	—
`schema-name`	Para inquilinos MySQL: el nombre de la base de datos. Para inquilinos Oracle: el nombre del propietario.	Sí	STRING	—	—
`table-name`	El nombre de la tabla de destino.	Sí	STRING	—	—
`parallel`	La concurrencia del lado del servidor para la tarea de importación. El servidor limita el grado real de paralelismo según las especificaciones de CPU del inquilino sin devolver un error. Fórmula: `MIN(tenant_cores × 2, parallel) × partition_nodes`. Por ejemplo, con 2 núcleos de CPU, `parallel=10` y 2 nodos de partición: `MIN(4, 10) × 2 = 8`.	No	INTEGER	`8`	—
`buffer-size`	El número de registros almacenados en búfer antes de una sola escritura en OceanBase.	No	INTEGER	`1024`	—
`dup-action`	El comportamiento cuando se encuentran claves primarias duplicadas. `STOP_ON_DUP`: falla la importación. `REPLACE`: sobrescribe la fila existente. `IGNORE`: descarta la fila entrante.	No	STRING	`REPLACE`	—
`load-method`	El modo de importación. `full`: importación directa estándar. `inc`: modo incremental, comprueba conflictos de clave primaria (observer 4.3.2+, `dup-action=REPLACE` no compatible). `inc_replace`: modo de reemplazo incremental, sobrescribe filas existentes directamente sin comprobaciones de conflictos (observer 4.3.2+, `dup-action` se ignora).	No	STRING	`full`	—
`max-error-rows`	El número máximo de filas con error toleradas. Las filas con error incluyen: claves primarias duplicadas cuando `dup-action=STOP_ON_DUP`, recuentos de columnas no coincidentes y filas donde falla la conversión de tipos.	No	LONG	`0`	—
`timeout`	El tiempo de expiración global para la tarea de importación directa.	No	DURATION	`7d`	—
`heartbeat-timeout`	El tiempo de expiración del latido del lado del cliente.	No	DURATION	`60s`	—
`heartbeat-interval`	El intervalo de latido del lado del cliente.	No	DURATION	`10s`	—

Mapeo de tipos

Modo compatible con MySQL

Tipo de OceanBase	Tipo de Flink
`TINYINT`	`TINYINT`
`SMALLINT`, `TINYINT UNSIGNED`	`SMALLINT`
`INT`, `MEDIUMINT`, `SMALLINT UNSIGNED`	`INT`
`BIGINT`, `INT UNSIGNED`	`BIGINT`
`BIGINT UNSIGNED`	`DECIMAL(20, 0)`
`REAL`, `FLOAT`	`FLOAT`
`DOUBLE`	`DOUBLE`
`NUMERIC(p, s)`, `DECIMAL(p, s)`	`DECIMAL(p, s)` (p ≤ 38)
`BOOLEAN`, `TINYINT(1)`	`BOOLEAN`
`DATE`	`DATE`
`TIME [(p)]`	`TIME [(p)] [WITHOUT TIME ZONE]`
`DATETIME [(p)]`, `TIMESTAMP [(p)]`	`TIMESTAMP [(p)] [WITHOUT TIME ZONE]`
`CHAR(n)`	`CHAR(n)`
`VARCHAR(n)`	`VARCHAR(n)`
`BIT(n)`	`BINARY(⌈n/8⌉)`
`BINARY(n)`	`BINARY(n)`
`VARBINARY(N)`	`VARBINARY(N)`
`TINYTEXT`, `TEXT`, `MEDIUMTEXT`, `LONGTEXT`	`STRING`
`TINYBLOB`, `BLOB`, `MEDIUMBLOB`, `LONGBLOB`	`BYTES` (máx. 2,147,483,647 bytes)

Modo compatible con Oracle

Tipo de OceanBase	Tipo de Flink
`NUMBER(p, s≤0)`, p−s < 3	`TINYINT`
`NUMBER(p, s≤0)`, p−s < 5	`SMALLINT`
`NUMBER(p, s≤0)`, p−s < 10	`INT`
`NUMBER(p, s≤0)`, p−s < 19	`BIGINT`
`NUMBER(p, s≤0)`, 19 ≤ p−s ≤ 38	`DECIMAL(p−s, 0)`
`NUMBER(p, s>0)`	`DECIMAL(p, s)`
`NUMBER(p, s≤0)`, p−s > 38	`STRING`
`FLOAT`, `BINARY_FLOAT`	`FLOAT`
`BINARY_DOUBLE`	`DOUBLE`
`NUMBER(1)`	`BOOLEAN`
`DATE`, `TIMESTAMP [(p)]`	`TIMESTAMP [(p)] [WITHOUT TIME ZONE]`
`CHAR(n)`, `NCHAR(n)`, `NVARCHAR2(n)`, `VARCHAR(n)`, `VARCHAR2(n)`, `CLOB`	`STRING`
`BLOB`, `ROWID`	`BYTES`

Ejemplos

Tabla de origen y tabla de destino

El siguiente ejemplo lee datos CDC de una tabla de origen OceanBase y los escribe en una tabla de destino JDBC. También se incluye una definición de tabla de destino de importación directa como referencia.

Las tres tablas utilizan 'connector' = 'oceanbase'. La tabla de origen y la de destino JDBC usan conjuntos de parámetros diferentes; la tabla de destino de carga directa establece sink.mode = 'direct-load' y se conecta mediante el puerto RPC.

-- OceanBase CDC source table (reads full history, then Binlog)
CREATE TEMPORARY TABLE oceanbase_source (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'     = 'oceanbase',
  'hostname'      = '<your-hostname>',
  'port'          = '3306',
  'username'      = '<your-username>',
  'password'      = '<your-password>',
  'database-name' = '<your-database-name>',
  'table-name'    = '<your-table-name>'
);

-- OceanBase JDBC sink table (for real-time streaming writes)
CREATE TEMPORARY TABLE oceanbase_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '<your-password>',
  'tableName' = '<your-table-name>'
);

-- OceanBase bypass import sink table (for high-throughput batch writes)
-- Requires a bounded data source; set Flink to batch mode for best performance
CREATE TEMPORARY TABLE oceanbase_directload_sink (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector'   = 'oceanbase',
  'sink.mode'   = 'direct-load',
  'host'        = '<your-host>',
  'port'        = '<your-rpc-port>',
  'tenant-name' = '<your-tenant-name>',
  'schema-name' = '<your-schema-name>',
  'table-name'  = '<your-table-name>',
  'username'    = '<your-username>',
  'password'    = '<your-password>'
);

BEGIN STATEMENT SET;
INSERT INTO oceanbase_sink
SELECT * FROM oceanbase_source;
END;

Tabla de dimensión

El siguiente ejemplo une un origen Datagen con una tabla de dimensión OceanBase mediante una unión temporal. La directiva de caché ALL carga la tabla de dimensión completa en memoria antes de que comience el trabajo.

CREATE TEMPORARY TABLE datagen_source (
  a INT,
  b BIGINT,
  c STRING,
  `proctime` AS PROCTIME()
) WITH (
  'connector' = 'datagen'
);

-- OceanBase dimension table with ALL cache policy
-- ALL cache loads the full table at startup — suitable for small, stable tables
CREATE TEMPORARY TABLE oceanbase_dim (
  a INT,
  b VARCHAR,
  c VARCHAR
) WITH (
  'connector' = 'oceanbase',
  'url'       = '<your-jdbc-url>',
  'userName'  = '<your-username>',
  'password'  = '${secret_values.password}',
  'tableName' = '<your-table-name>'
);

CREATE TEMPORARY TABLE blackhole_sink (
  a INT,
  b STRING
) WITH (
  'connector' = 'blackhole'
);

INSERT INTO blackhole_sink
SELECT T.a, H.b
FROM datagen_source AS T
JOIN oceanbase_dim FOR SYSTEM_TIME AS OF T.`proctime` AS H
ON T.a = H.a;

Pasos siguientes

Supported connectors — lista completa de conectores disponibles en Realtime Compute for Apache Flink
Descripción general del servicio Binlog de OceanBase — necesario para lecturas incrementales CDC