SelectDB (Beta) - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

El conector de SelectDB integra Realtime Compute for Apache Flink con ApsaraDB for SelectDB, un almacén de datos en tiempo real totalmente administrado y compatible con Apache Doris en Alibaba Cloud. Úselo para construir pipelines en tiempo real que lean, escriban o busquen datos en SelectDB, así como para ejecutar la sincronización completa de bases de datos en trabajos de ingesta de datos basados en YAML.

Capacidades compatibles:

Categoría	Detalles
Tipos de tabla	Tabla de origen, tabla de destino, tabla de dimensiones, destino de ingesta de datos
Modo de ejecución	Streaming y por lotes
Formato de datos	JSON y CSV
Tipo de API	DataStream, SQL y trabajos YAML de ingesta de datos
Compatibilidad con actualizaciones y eliminaciones	Sí
Métricas de monitoreo	Ninguna

Características principales:

Sincronización completa de los datos de la base de datos.
Semántica exactly-once mediante confirmación en dos fases (2PC), sin registros duplicados ni perdidos.
Compatible con Apache Doris 1.0 y versiones posteriores.

Requisitos previos

Antes de empezar, asegúrese de contar con lo siguiente:

Realtime Compute for Apache Flink con Ververica Runtime (VVR) 8.0.10 o una versión posterior.
Una instancia de ApsaraDB for SelectDB. Consulte Create an instance.
Una lista blanca de direcciones IP configurada en la instancia. Consulte Configure a whitelist.

Configurar el conector

El conector de SelectDB viene integrado en VVR 11.1 y versiones posteriores, por lo que no se requiere instalación manual.

Para las versiones de VVR 8.0.10 a 11.0, instale el conector manualmente:

Descargue el paquete JAR desde Maven Central (versiones de Flink 1.15 a 1.17).
Cargue el archivo JAR en su consola de desarrollo de Realtime Compute for Apache Flink. Consulte Manage custom connectors.
Haga referencia al conector en su trabajo SQL mediante 'connector' = 'doris'.

SQL

Sintaxis

Los tres tipos de tabla (origen, destino y dimensiones) comparten la misma sintaxis DDL. El rol de la tabla se define según los parámetros que incluya.

Para usar SelectDB como tabla de origen, primero habilite la conexión directa al clúster. En la consola de ApsaraDB for SelectDB, vaya a Instance Details > Network Information y haga clic en Enable Direct Cluster Connection . Esta acción activa el protocolo Arrow Flight SQL para lecturas paralelas de alto rendimiento.

CREATE TABLE selectdb_source (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING
) WITH (
  'connector'        = 'doris',
  'fenodes'          = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username'         = 'admin',
  'password'         = '****'
);

Parámetros

Generales

Parámetro	Obligatorio	Predeterminado	Descripción
`connector`	Sí	—	Valor fijo: `doris`.
`fenodes`	Sí	—	Endpoint HTTP de la instancia de SelectDB: `<VPC Address or Public Address>:<HTTP Protocol Port>`. Obtenga ambos valores en Instance Details > Network Information dentro de la consola de SelectDB. Ejemplo: `selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080`.
`jdbc-url`	No	—	Cadena de conexión Java Database Connectivity (JDBC) para búsquedas en tablas de dimensiones y consultas de metadatos: `jdbc:mysql://<VPC Address or Public Address>:<MySQL Protocol Port>`. Ejemplo: `jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030`.
`table.identifier`	Sí	—	Tabla de destino con el formato `<database>.<table>`. Ejemplo: `db.tbl`.
`username`	Sí	—	Nombre de usuario de la base de datos. Si es necesario, restablezca la contraseña desde la esquina superior derecha de la página Instance Details.
`password`	Sí	—	Contraseña del usuario de la base de datos.
`doris.request.retries`	No	`3`	Número de reintentos para solicitudes con errores.
`doris.request.connect.timeout`	No	`30s`	Tiempo de espera de conexión.
`doris.request.read.timeout`	No	`30s`	Tiempo de espera de lectura.

Tabla de origen

Parámetro	Obligatorio	Predeterminado	Descripción
`doris.request.query.timeout`	No	`21600s`	Tiempo de espera de la consulta (6 horas de forma predeterminada).
`doris.request.tablet.size`	No	`1`	Número de tablets por partición. Los valores más bajos aumentan el paralelismo de Flink, pero generan mayor presión sobre la base de datos.
`doris.batch.size`	No	`4064`	Máximo de filas leídas desde un nodo Backend (BE) por solicitud. Increméntelo para reducir la sobrecarga de conexión y la latencia de red.
`doris.exec.mem.limit`	No	`8192mb`	Límite de memoria por consulta en bytes (8 GB de forma predeterminada).
`source.use-flight-sql`	No	`false`	No requiere configuración adicional. Al habilitar Direct Cluster Connection en la consola de SelectDB, Arrow Flight SQL se activa automáticamente.
`source.flight-sql-port`	No	—	Puerto de Arrow Flight SQL (`arrow_flight_sql_port`) del nodo Frontend (FE).

Tabla de destino

El modo de escritura afecta las garantías de entrega y el comportamiento de vaciado. Elija la opción adecuada según sus requisitos de consistencia:

	Escritura en streaming	Escritura por lotes
Condición de activación	Sigue los intervalos de checkpoint de Flink	Vaciado periódico por volumen de datos o umbral de tiempo
Garantía de entrega	Exactly-once (mediante 2PC)	At-least-once; logre idempotencia con el modelo Unique
Latencia	Limitada por el intervalo de checkpoint	Flexible e independiente de los checkpoints
Tolerancia a fallos	Recuperación completa del estado de Flink	Depende de la deduplicación del modelo Unique

Parámetro	Obligatorio	Predeterminado	Descripción
`sink.label-prefix`	No	—	Prefijo de etiqueta para importaciones de Stream Load. Debe ser único a nivel global entre todos los trabajos, ya que la misma etiqueta solo se puede confirmar una vez. Es obligatorio para garantizar la semántica exactly-once tras reiniciar los trabajos.
`sink.properties.*`	No	—	Parámetros de importación de Stream Load que se pasan directamente a la API de Stream Load de SelectDB. Consulte los ejemplos a continuación.
`sink.enable-delete`	No	`true`	Propaga las operaciones DELETE. Requiere que la tabla de Doris tenga habilitada la eliminación por lotes y solo funciona con el modelo Unique.
`sink.enable-2pc`	No	`true`	Habilita la confirmación en dos fases (2PC) para lograr la semántica exactly-once. Consulte Explicit Transaction Operations.
`sink.buffer-size`	No	`1 MB`	Tamaño del búfer de caché de escritura en bytes. Mantenga el valor predeterminado.
`sink.buffer-count`	No	`3`	Cantidad de búferes de caché de escritura. Mantenga el valor predeterminado.
`sink.max-retries`	No	`3`	Máximo de reintentos después de un error de confirmación.
`sink.enable.batch-mode`	No	`false`	Cambia al modo de escritura por lotes. El vaciado se controla mediante los tres parámetros `sink.buffer-flush.*` que se muestran a continuación, en lugar de los checkpoints. No garantiza la semántica exactly-once; utilice el modelo Unique para lograr idempotencia.
`sink.flush.queue-size`	No	`2`	Tamaño de la cola de caché en modo por lotes.
`sink.buffer-flush.max-rows`	No	`500000`	Máximo de filas por vaciado en modo por lotes.
`sink.buffer-flush.max-bytes`	No	`100 MB`	Máximo de bytes por vaciado en modo por lotes.
`sink.buffer-flush.interval`	No	`10s`	Intervalo de vaciado en modo por lotes.
`sink.ignore.update-before`	No	`true`	Omite los eventos update-before provenientes de Flink CDC.

**Ejemplos de sink.properties.*:**

Formato CSV:

'sink.properties.column_separator' = ','
-- If values may contain commas, use a non-printable separator:
-- 'sink.properties.column_separator' = '\x01'

Formato JSON:

'sink.properties.format'            = 'json',
'sink.properties.read_json_by_line' = 'true'
-- Alternatively: 'sink.properties.strip_outer_array' = 'true'

Tabla de dimensiones

Parámetro	Obligatorio	Predeterminado	Descripción
`lookup.cache.max-rows`	No	`-1`	Máximo de filas en la caché de búsqueda. El valor `-1` deshabilita la caché.
`lookup.cache.ttl`	No	`10s`	Tiempo de vida (TTL) de las entradas de la caché.
`lookup.max-retries`	No	`1`	Reintentos tras fallar una consulta de búsqueda.
`lookup.jdbc.async`	No	`false`	Habilita la búsqueda asíncrona.
`lookup.jdbc.read.batch.size`	No	`128`	Tamaño máximo de lote por consulta en modo de búsqueda asíncrona.
`lookup.jdbc.read.batch.queue-size`	No	`256`	Tamaño de la cola del búfer intermedio en modo de búsqueda asíncrona.
`lookup.jdbc.read.thread-size`	No	`3`	Hilos de búsqueda JDBC por tarea en modo de búsqueda asíncrona.

Ejemplos

Tabla de origen

CREATE TEMPORARY TABLE selectdb_source (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING
) WITH (
  'connector'        = 'doris',
  'fenodes'          = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username'         = 'admin',
  'password'         = '****'
);

Tabla de destino

CREATE TEMPORARY TABLE selectdb_sink (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING
) WITH (
  'connector'        = 'doris',
  'fenodes'          = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username'         = 'admin',
  'password'         = '****',
  'sink.label-prefix' = 'flink_orders'  -- Must be globally unique across jobs
);

Tabla de dimensiones

SelectDB actúa como tabla de dimensiones de búsqueda combinada con una tabla de hechos en streaming.

-- Fact table from Kafka
CREATE TEMPORARY TABLE fact_table (
  `id`           BIGINT,
  `name`         STRING,
  `city`         STRING,
  `process_time` AS proctime()
) WITH (
  'connector' = 'kafka',
  ...
);

-- Dimension table from SelectDB
CREATE TEMPORARY TABLE dim_city (
  `city`     STRING,
  `level`    INT,
  `province` STRING,
  `country`  STRING
) WITH (
  'connector'        = 'doris',
  'fenodes'          = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'jdbc-url'         = 'jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030',
  'table.identifier' = 'dim.dim_city',
  'username'         = 'admin',
  'password'         = '****'
);

-- Temporal join
SELECT a.id, a.name, a.city, c.province, c.country, c.level
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city;

Ingesta de datos

Utilice el conector de SelectDB como destino en trabajos de ingesta de datos basados en YAML para sincronizar bases de datos completas.

Sintaxis

source:
  type: <source-type>

sink:
  type: doris
  name: Doris Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  username: root
  password: ""

Parámetros

Parámetro	Obligatorio	Predeterminado	Descripción
`type`	Sí	—	Valor fijo: `doris`.
`name`	No	—	Nombre descriptivo del destino.
`fenodes`	Sí	—	Endpoint HTTP: `<VPC Address or Public Address>:<HTTP Protocol Port>`. Obtenga ambos valores en Instance Details > Network Information dentro de la consola de SelectDB. Ejemplo: `selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080`.
`jdbc-url`	No	—	Cadena de conexión JDBC. Ejemplo: `jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030`.
`username`	Sí	—	Nombre de usuario de la base de datos.
`password`	Sí	—	Contraseña del usuario de la base de datos.
`sink.enable.batch-mode`	No	`true`	El modo por lotes está habilitado de forma predeterminada en los trabajos de ingesta de datos. El vaciado se controla mediante los tres parámetros `sink.buffer-flush.*`. No garantiza la semántica exactly-once; utilice el modelo Unique para lograr idempotencia.
`sink.flush.queue-size`	No	`2`	Tamaño de la cola de caché.
`sink.buffer-flush.max-rows`	No	`500000`	Máximo de filas por vaciado.
`sink.buffer-flush.max-bytes`	No	`100 MB`	Máximo de bytes por vaciado.
`sink.buffer-flush.interval`	No	`10s`	Intervalo de vaciado. Mínimo: `1s`.
`sink.properties.*`	No	—	Parámetros de importación de Stream Load.

**Ejemplos de sink.properties.*:**

Formato CSV:

sink.properties.column_separator: ','
# If values may contain commas, use a non-printable separator:
# sink.properties.column_separator: '\x01'

Formato JSON:

sink.properties.format: 'json'
sink.properties.read_json_by_line: 'true'

Ejemplos

A continuación se presentan ejemplos de escenarios comunes de ingesta de datos. Sustituya los endpoints, nombres de usuario y contraseñas de marcador de posición por sus valores reales.

La expresión ${secret_values.variable_name} hace referencia a una variable secreta creada en su espacio de trabajo. Para obtener más información, consulte Manage variables. Para ver la descripción completa de los parámetros de origen, consulte MySQL YAML connector y Postgres CDC connector.

Nota

El conector no crea las bases de datos de destino. Antes de empezar, cree en su instancia de ApsaraDB for SelectDB las bases de datos de destino que se usan en los siguientes ejemplos. Las tablas de destino se crean automáticamente si no existen.

Sincronización de una sola tabla

Sincronice una única tabla de MySQL hacia SelectDB. La tabla de destino se crea automáticamente si no existe.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  # (Optional) Synchronize the existing full data first, and then the incremental data.
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

Sincronización completa de la base de datos

Sincronice todas las tablas de una base de datos MySQL hacia SelectDB. Los nombres de las bases de datos y tablas descendentes coinciden con los de origen. Las tablas de destino se crean automáticamente si no existen.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # Matches all tables in the test_db database.
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # (Optional) Synchronize tables that are created during the incremental phase without restarting the job.
  scan.binlog.newly-added-table.enabled: true

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # (Optional) The flush interval for batch writing. Lower this value for small workloads so that data is not held in the buffer for too long.
  sink.buffer-flush.interval: 10s

pipeline:
  name: MySQL to SelectDB Pipeline

Exclusión de tablas en la sincronización completa

Omita las tablas que no desea sincronizar, como las temporales o las de registro. Sin un módulo de enrutamiento, los datos se escriben en la tabla de destino que tiene el mismo nombre que la tabla de origen.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  # Tables that match this regular expression are not synchronized. Separate multiple regular expressions with commas (,).
  tables.exclude: test_db.tmp_\.*, test_db.log_\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

Sincronización hacia bases de datos y tablas específicas

Utilice el módulo de enrutamiento para renombrar los destinos cuando los nombres de las bases de datos o tablas descendentes difieren de los originales. El siguiente ejemplo sincroniza las tablas de test_db hacia la base de datos ods_db y añade el prefijo ods_ al nombre de cada tabla. Por ejemplo, test_db.orders se sincroniza como ods_db.ods_orders.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

route:
  # <> is a placeholder that is replaced with the matched source table name.
  - source-table: test_db.\.*
    sink-table: ods_db.ods_<>
    replace-symbol: <>

pipeline:
  name: MySQL to SelectDB Pipeline

Fusión de tablas fragmentadas

Combine varias tablas de fragmentos que comparten el mismo esquema en una única tabla de SelectDB. El siguiente ejemplo utiliza el módulo transform para agregar una columna que identifica la base de datos y la tabla de origen. Después, usa esta columna junto con la clave principal de negocio como clave principal compuesta, lo que permite rastrear el origen de los datos fusionados.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # Matches all shards, such as order_db_1.orders_1 and order_db_2.orders_2.
  tables: order_db_\d+.orders_\d+
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # Set this parameter to false when the downstream primary key differs from the upstream one. Otherwise, the rows that correspond to the old primary key are not deleted.
  sink.ignore.update-before: false

transform:
  - source-table: order_db_\d+.orders_\d+
    # Append the src_table column that identifies the source database and table.
    projection: "*, __schema_name__ || '.' || __table_name__ AS src_table"
    # The composite primary key must be the same as the key columns of the destination table.
    primary-keys: order_id, src_table
    description: Append the source identifier and set a composite primary key

route:
  # Merge all shards into a single table.
  - source-table: order_db_\d+.orders_\d+
    sink-table: dw_db.merged_orders

pipeline:
  name: MySQL sharding to SelectDB Pipeline

Filtrado de datos y reducción de columnas

Utilice el módulo transform para filtrar datos y reducir columnas. La condición de filtro se evalúa sobre los campos de la tabla de origen para determinar si un registro se sincroniza, mientras que la proyección define qué columnas se escriben en la tabla descendente.

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

transform:
  - source-table: test_db.test_source_table
    # Synchronize only the records whose order_status is PAID.
    filter: "order_status = 'PAID'"
    # Write only the following four columns.
    projection: order_id, customer_id, total_amount, created_at
    primary-keys: order_id
    description: Filter by order status and prune columns

pipeline:
  name: MySQL to SelectDB Pipeline

Sincronización completa de una base de datos PostgreSQL

Sincronice todas las tablas de un esquema de PostgreSQL hacia SelectDB. Antes de empezar, establezca wal_level en logical en la instancia de PostgreSQL y compruebe que max_replication_slots y max_wal_senders tengan capacidad suficiente. Para obtener más información, consulte Configure a PostgreSQL database.

source:
  type: postgres
  name: PostgreSQL Source
  hostname: <yourHostname>
  port: 5432
  username: ${secret_values.pg_username}
  password: ${secret_values.pg_password}
  # PostgreSQL table names use the database.schema.table format.
  tables: test_db.public.\.*
  slot.name: <yourSlotName>
  decoding.plugin.name: pgoutput
  scan.startup.mode: initial
  # (Optional) Use an existing publication instead of creating one when the job starts.
  debezium.publication.autocreate.mode: disabled
  debezium.publication.name: <yourPublicationName>

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: PostgreSQL to SelectDB Pipeline

Mapeo de tipos

De Flink a SelectDB

Tipo de Flink CDC	Tipo de SelectDB	Notas
`TINYINT`	`TINYINT`
`SMALLINT`	`SMALLINT`
`INT`	`INT`
`BIGINT`	`BIGINT`
`DECIMAL`	`DECIMAL`
`FLOAT`	`FLOAT`
`DOUBLE`	`DOUBLE`
`BOOLEAN`	`BOOLEAN`
`DATE`	`DATE`
`TIMESTAMP[(p)]`	`DATETIME[(p)]`
`TIMESTAMP_LTZ[(p)]`	`DATETIME[(p)]`
`CHAR(n)`	`CHAR(n*3)`	SelectDB almacena cadenas en UTF-8. Los caracteres en inglés ocupan 1 byte y los caracteres chinos ocupan 3 bytes. La longitud máxima de `CHAR` es 255; los valores más largos se convierten automáticamente a `VARCHAR`.
`VARCHAR(n)`	`VARCHAR(n*3)`	Aplica el mismo multiplicador de UTF-8. La longitud máxima de `VARCHAR` es 65533; los valores más largos se convierten automáticamente a `STRING`.
`BINARY(n)`	`STRING`
`VARBINARY(n)`	`STRING`
`STRING`	`STRING`

Realtime Compute for Apache Flink:SelectDB

Requisitos previos

Configurar el conector

SQL

Sintaxis

Parámetros

Generales

Tabla de origen

Tabla de destino

Tabla de dimensiones

Ejemplos

Tabla de origen

Tabla de destino

Tabla de dimensiones

Ingesta de datos

Sintaxis

Parámetros

Ejemplos

Sincronización de una sola tabla

Sincronización completa de la base de datos

Exclusión de tablas en la sincronización completa

Sincronización hacia bases de datos y tablas específicas

Fusión de tablas fragmentadas

Filtrado de datos y reducción de columnas

Sincronización completa de una base de datos PostgreSQL

Mapeo de tipos

De Flink a SelectDB

Pasos siguientes