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

Este tema describe cómo crear y gestionar tablas de Fluss mediante Flink SQL.

Limitaciones

Esta función solo está disponible en Realtime Compute for Apache Flink VVR 11.2 y versiones posteriores.

Crear una tabla con clave primaria

Inicie sesión en la consola de Realtime Compute for Apache Flink.
En la columna Actions del espacio de trabajo deseado, haga clic en Console.
En el panel de navegación izquierdo, elija Development > Scripts > New Temporary Script.
Escriba y ejecute la sentencia SQL.

CREATE TABLE `my-catalog`.`my_db`.`my_pk_table` (
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (shop_id, user_id) NOT ENFORCED
) WITH (
  'bucket.num' = '4'
);

Nota

Debe planificar el número de buckets al crear una tabla.

Aunque es posible ajustar la distribución en buckets de una tabla de registros, el número de buckets de una tabla con clave primaria no se puede modificar después de su creación. Por lo tanto, debe determinar el número de buckets según el volumen de datos al crear la tabla. Utilice la siguiente fórmula:

Número de buckets = Volumen total de datos en una sola partición / Tamaño objetivo del bucket

Crear una tabla de registros

CREATE TABLE `my-catalog`.`my_db`.`my_log_table` (
  order_id BIGINT,
  item_id BIGINT,
  amount INT,
  address STRING
) WITH (
  'bucket.num' = '8'
);

Crear una tabla particionada

Actualmente, Fluss solo admite el tipo de dato STRING para las claves de partición.
En una tabla con clave primaria particionada, la clave de partición (el campo dt en el siguiente ejemplo) debe ser un subconjunto de la clave primaria.

Tabla con clave primaria particionada

La clave de partición es dt y la clave primaria es [dt, shop_id, user_id].

CREATE TABLE `my-catalog`.`my_db`.`my_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

Tabla de registros particionada

El siguiente ejemplo crea una tabla de registros particionada con dt como clave de partición.

CREATE TABLE `my-catalog`.`my_db`.`my_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
);

Agregar una partición

ALTER TABLE my-catalog.my_db.my_part_pk_table 
ADD PARTITION (dt = '2025-03-05');

Crear una tabla con particionamiento automático

Una tabla con particionamiento automático extiende una tabla particionada mediante la creación automática de particiones. Fluss crea las particiones de forma anticipada según su directiva de particionamiento automático.

Tabla con clave primaria y particionamiento automático

La clave de partición es dt y el intervalo de particionamiento automático es diario.

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_pk_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT,
  PRIMARY KEY (dt, shop_id, user_id) NOT ENFORCED
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

Tabla de registros con particionamiento automático

La clave de partición es dt y el intervalo de particionamiento automático es diario.

CREATE TABLE `my-catalog`.`my_db`.`my_auto_part_log_table` (
  dt STRING,
  shop_id BIGINT,
  user_id BIGINT,
  num_orders INT,
  total_amount INT
) PARTITIONED BY (dt) WITH (
  'bucket.num' = '4'
  'table.auto-partition.enabled' = 'true',
  'table.auto-partition.time-unit' = 'day'
);

Actualizar una tabla

Agregar columnas

Nota

Esta función requiere Realtime Compute for Apache Flink VVR 11.5 o posterior y Fluss 0.8-ali-3.0 o posterior. Para evitar interrumpir las operaciones de lectura y escritura, primero actualice su trabajo de Flink a VVR 11.5 o posterior y luego actualice su clúster de Fluss a 0.8-ali-3.0.

Fluss permite evolucionar el esquema de una tabla mediante la adición de nuevas columnas. Puede agregar columnas de cualquier tipo de dato, incluidos tipos complejos como ROW, MAP y ARRAY.

Se trata de una operación ligera que solo afecta a los metadatos y ofrece los siguientes beneficios:

Sin reescritura de datos: No es necesario reescribir ni migrar los archivos de datos existentes al agregar columnas.
Ejecución instantánea: La operación se completa en milisegundos, independientemente del tamaño de la tabla.
Alta disponibilidad: La tabla permanece en línea y totalmente disponible durante la evolución del esquema, sin interrupciones para los clientes en ejecución.

Actualmente, esta función presenta las siguientes limitaciones:

Orden de columnas: Las nuevas columnas siempre se agregan al final de la lista de columnas existente.
Valores nulos: Solo se pueden agregar columnas que admitan valores nulos a una tabla para garantizar la compatibilidad con los datos existentes.
Campos anidados: Actualmente no se admite la adición de campos a un tipo ROW anidado existente. Este tipo de operación se clasifica como "actualización de tipos de columna" y será compatible en una versión futura.

Utilice la sentencia ALTER TABLE para agregar una sola columna o varias columnas.

-- Add a single column at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD user_email STRING COMMENT 'User email address';

-- Add multiple columns at the end of the table
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` ADD (
    user_email STRING COMMENT 'User email address',
    order_quantity INT
);

Modificar propiedades de la tabla

Utilice la sentencia ALTER TABLE SET para modificar una o más opciones de conector y opciones de almacenamiento. Los nuevos valores sobrescriben los existentes.

El clúster de Fluss aplica los cambios en las opciones de almacenamiento de forma dinámica. No es necesario reconstruir la tabla.

Opciones compatibles

Todas las opciones de lectura, escritura, búsqueda y otras opciones, excepto bootstrap.servers.
Las siguientes opciones de almacenamiento:
- table.datalake.enabled: Habilita o deshabilita la integración de data lake y streaming.
- table.datalake.freshness: Establece la actualización de datos para el almacenamiento de data lake.

-- Enable data lake and stream integration
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.enabled' = 'true');

-- Set data lake freshness to 5 minutes
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` SET ('table.datalake.freshness' = '5min');

Importante

Una vez habilitada la integración de data lake y streaming (table.datalake.enabled) en una tabla, ya no se pueden modificar las opciones de prefijo de formato de lago, como paimon.*.

Restablecer propiedades de la tabla

Utilice la sentencia ALTER TABLE RESET para restaurar una o más opciones de conector y opciones de almacenamiento a sus valores predeterminados.

-- Reset data lake and stream integration to its default (disabled)
ALTER TABLE `my-catalog`.`my_db`.`my_pk_table` RESET ('table.datalake.enabled');

Clonar una tabla (clonación de esquema)

La sintaxis CREATE TABLE LIKE crea una nueva tabla con el mismo esquema, información de partición y propiedades que una tabla existente.

Debe ejecutar esta operación en un script de consulta temporal. Para abrir un nuevo script, elija Development > Scripts > New Temporary Script.

-- Create a temporary table named datagen
CREATE TEMPORARY TABLE datagen (
 user_id BIGINT,
 item_id BIGINT,
 behavior STRING,
 dt STRING,
 hh STRING
) WITH (
 'connector' = 'datagen',
 'rows-per-second' = '10'
);
-- Create a Fluss table based on the datagen table, excluding its options
CREATE TABLE my-catalog.my_db.my_like_db LIKE datagen (EXCLUDING OPTIONS);

Eliminar una tabla

DROP TABLE `my-catalog`.`my_db`.`my_pk_table`;