Realtime Compute for Apache Flink:Gestionar catálogos de Iceberg

Requisitos previos

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

• Un espacio de trabajo de Flink que ejecute Ververica Runtime (VVR) 11.1 o posterior

• Un catálogo DLF en la misma región que su espacio de trabajo de Flink. Consulte Quick start with DLF

Limitaciones

Crear un catálogo DLF de Iceberg

Al crear un catálogo Iceberg en Flink, se registra una asignación hacia su catálogo DLF existente. Esta acción no crea ni modifica datos en DLF. Todas las tablas creadas en el catálogo DLF mediante Iceberg REST son tablas Iceberg.

1. Inicie sesión en la consola de Realtime Compute for Apache Flink.

2. En la columna Actions de su espacio de trabajo, haga clic en Console.

3. En el menú de navegación izquierdo, vaya a Development > Scripts y cree un nuevo script.

4. En el editor SQL, pegue la siguiente sentencia. En la esquina inferior derecha, haga clic en Environment y seleccione un clúster de sesiones que ejecute VVR 11.2.0 o posterior. Luego, ejecute la sentencia.

   CREATE CATALOG `catalog_name`
   WITH (
       'type'              = 'iceberg',
       'catalog-type'      = 'rest',
       'token.provider'    = 'dlf',
       'uri'               = 'http://cn-hangzhou-vpc.dlf.aliyuncs.com/iceberg',
       'warehouse'         = 'iceberg_test',
       'rest.signing-region' = 'cn-hangzhou',
       'io-impl'           = 'org.apache.iceberg.rest.DlfFileIO'
   );

   Reemplace los valores de uri, warehouse y rest.signing-region con los suyos propios. Consulte Endpoint REST de Iceberg para ver la lista de URI disponibles por región. La siguiente tabla describe cada opción.

   Opción	Obligatorio	Descripción	Ejemplo
   type	Sí	Establezca este valor en iceberg.	iceberg
   catalog-type	Sí	Establezca este valor en rest.	rest
   token.provider	Sí	Establezca este valor en dlf. Configura la autenticación basada en DLF.	dlf
   uri	Sí	El endpoint REST de Iceberg para su catálogo DLF. La URI varía según la región. Consulte Endpoint REST de Iceberg.	http://ap-southeast-1-vpc.dlf.aliyuncs.com/iceberg
   warehouse	Sí	El nombre de su catálogo DLF.	iceberg_test
   rest.signing-region	Sí	El ID de la región donde se despliega DLF.	cn-hangzhou
   io-impl	Sí	Establezca este valor en org.apache.iceberg.rest.DlfFileIO.	org.apache.iceberg.rest.DlfFileIO

Consultar o eliminar un catálogo

1. En la consola de Realtime Compute for Apache Flink, haga clic en Console en la columna Actions de su espacio de trabajo.

2. Vaya a la página Data Management y abra Catalog List.

   • Para consultar las bases de datos y tablas de un catálogo, haga clic en View junto al nombre del catálogo.

   • Para eliminar un catálogo, haga clic en Delete en la columna Actions.

     Importante

     La eliminación de un catálogo solo quita su registro de Data Management en su proyecto de Flink. Los archivos de datos subyacentes de Iceberg en DLF no se ven afectados. Para reutilizar las tablas, registre el catálogo nuevamente.

   Como alternativa, ejecute la siguiente sentencia en la página Data Query:

   DROP CATALOG <catalog_name>;

Gestionar bases de datos Iceberg

Ejecute las siguientes sentencias en la página Data Query: seleccione el código y haga clic en Run.

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Replace my_db with the database name you want to create.
CREATE DATABASE `my_db`;

-- Replace my-catalog with your catalog name.
USE CATALOG `my-catalog`;

-- Delete a database only if it is empty.
DROP DATABASE `my_db`;

-- Delete a database and all tables inside it.
DROP DATABASE `my_db` CASCADE;

Gestionar tablas Iceberg

Una vez registrado un catálogo Iceberg, puede hacer referencia a sus tablas en un trabajo sin declarar lenguaje de definición de datos (DDL). Utilice el nombre completo de tres partes <catalog>.<database>.<table>, o bien ejecute primero USE CATALOG y USE <database>, y luego haga referencia a la tabla solo por su nombre.

-- Replace my-catalog with your Iceberg catalog name.
-- Replace my_db with your database name.
-- Replace my_tbl with your table name.
CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
  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);

ALTER TABLE my_table RENAME TO my_table_new;

-- Add c1 (INT) and c2 (STRING) to the end of the table.
ALTER TABLE my_table ADD (c1 INT, c2 STRING);

-- Rename column c0 to c1.
ALTER TABLE my_table RENAME c0 TO c1;

-- Delete columns c1 and c2.
ALTER TABLE my_table DROP (c1, c2);

ALTER TABLE my_table MODIFY buy_count BIGINT COMMENT 'this is buy count';

-- Move col_a to the beginning of the table.
ALTER TABLE my_table MODIFY col_a DOUBLE FIRST;

-- Move col_a immediately after col_b.
ALTER TABLE my_table MODIFY col_a DOUBLE AFTER col_b;

-- Temporarily enable upsert mode when writing to a table.
INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
SELECT ...;

-- Temporarily set the source read interval to 10 seconds.
SELECT * FROM my_table /*+ OPTIONS('monitor-interval'='10s') */;

-- Replace my-catalog, my_db, and my_tbl with your values.
DROP TABLE `my-catalog`.`my_db`.`my_tbl`;