Realtime Compute for Apache Flink:Inicio rápido de Paimon: funciones básicas

Requisitos previos

• Si utiliza un usuario de RAM o un rol de RAM, asegúrese de que tenga los permisos obligatorios para acceder a la consola de Realtime Compute for Apache Flink. Para obtener más información, consulte Permission management.

• Ya creó un espacio de trabajo de Flink. Para obtener más información, consulte Activate Realtime Compute for Apache Flink.

• Activó Object Storage Service (OSS) y creó un bucket con la clase de almacenamiento Standard. Para obtener más información, consulte Get started by using the OSS console. OSS se usa para almacenar los archivos de las tablas de Paimon, incluidos los archivos de datos y de metadatos.

• Las tablas de Paimon solo son compatibles con Ververica Runtime (VVR) 8.0.5 o versiones posteriores.

Paso 1: Crear un catálogo de Paimon

1. Acceda al editor de scripts.

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

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

   3. En la barra de navegación izquierda, haga clic en Data Development > Data Query. En la pestaña Query Script, cree un nuevo Query Script.

2. En el editor, introduzca el siguiente código para crear un catálogo de Paimon.

   -- my-catalog is a custom catalog name.
   CREATE Catalog `my-catalog` WITH (
     'type' = 'paimon',
     'metastore' = 'filesystem',
     'warehouse' = '<warehouse>',
     'fs.oss.endpoint' = '<fs.oss.endpoint>',
     'fs.oss.accessKeyId' = '<fs.oss.accessKeyId>',
     'fs.oss.accessKeySecret' = '<fs.oss.accessKeySecret>'
   );

   A continuación se describen los parámetros.

   Parámetro	Descripción	Obligatorio	Observaciones
   type	El tipo de catálogo.	Sí	El valor debe ser 'paimon'.
   metastore	El tipo de metastore.	Sí	Este ejemplo usa filesystem. Para obtener más información sobre otros tipos, consulte Manage Paimon catalogs.
   warehouse	El directorio warehouse en su bucket de OSS.	Sí	El formato es oss://<bucket>/<object>, donde: bucket: el nombre del bucket de OSS que creó. object: la ruta donde desea almacenar los datos. Puede encontrar los nombres de su bucket y de su object en la consola de OSS.
   fs.oss.endpoint	El endpoint del servicio de OSS.	No	Obligatorio si el bucket de OSS especificado en warehouse está en una región distinta a la de su espacio de trabajo de Flink o pertenece a otra cuenta de Alibaba Cloud. Para obtener más información, consulte Regions and Endpoints. Nota Si desea almacenar una tabla de Paimon en OSS-HDFS, debe especificar los parámetros fs.oss.endpoint, fs.oss.accessKeyId y fs.oss.accessKeySecret. El valor de fs.oss.endpoint tiene el formato cn-<region>.oss-dls.aliyuncs.com, por ejemplo, cn-hangzhou.oss-dls.aliyuncs.com.
   fs.oss.accessKeyId	El AccessKey ID de una cuenta de Alibaba Cloud o de un usuario de RAM que tenga permisos de lectura y escritura en OSS.	No	Obligatorio si el bucket de OSS especificado en warehouse está en una región distinta a la de su espacio de trabajo de Flink o pertenece a otra cuenta de Alibaba Cloud. Para obtener información sobre cómo conseguir un par de AccessKey, consulte Create an AccessKey pair.
   fs.oss.accessKeySecret	El AccessKey secret de la cuenta de Alibaba Cloud o del usuario de RAM que tenga permisos de lectura y escritura en OSS.	No

3. Seleccione el código que crea el catálogo de Paimon y haga clic en Run a la izquierda.

   Un mensaje The following statement has been executed successfully! indica que el catálogo se creó correctamente.

Paso 2: Crear una base de datos y una tabla de Paimon

1. En la página Scripts, introduzca el siguiente código para crear una base de datos de Paimon llamada my_db y una tabla de Paimon llamada my_tbl.

   CREATE DATABASE `my-catalog`.`my_db`;
   CREATE TABLE `my-catalog`.`my_db`.`my_tbl` (
     dt STRING,
     id BIGINT,
     content STRING,
     PRIMARY KEY (dt, id) NOT ENFORCED
   ) PARTITIONED BY (dt) WITH (
     'changelog-producer' = 'lookup'  
   );

   Nota

   Para habilitar más adelante el consumo mediante streaming de la tabla de Paimon, este ejemplo establece changelog-producer en lookup dentro de la cláusula WITH. Esto usa la estrategia lookup para generar el registro de cambios. Para obtener más información, consulte Changelog production.

2. Seleccione el código que crea la base de datos y la tabla de Paimon, y luego haga clic en Run a la izquierda.

   Si se devuelve el mensaje The following statement has been executed successfully!, la base de datos my_db y la tabla my_tbl se crearon correctamente.

Paso 3: Escribir datos en la tabla de Paimon

1. En la pestaña Drafts de la página Development > ETL, cree un borrador de flujo en blanco. Para obtener más información, consulte Develop an SQL draft. Copie la siguiente instrucción INSERT en el editor SQL.

   -- Paimon commits data only after each checkpoint is complete.
   -- The checkpoint interval is shortened to 10s to commit data faster.
   -- In a production environment, typically set the checkpoint interval and the minimum interval between checkpoints from 1 to 10 minutes, based on your latency requirements.
   SET 'execution.checkpointing.interval'='10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108',1,'apple'), ('20240108',2,'banana'), ('20240109',1,'cat'), ('20240109',2,'dog');

2. En la esquina superior derecha del editor SQL, haga clic en Deploy. En el cuadro de diálogo Deploy draft, configure los parámetros según las indicaciones y haga clic en Confirm.

3. En la página O&M > Deployments, busque el despliegue correspondiente y haga clic en Start en la columna Actions. Seleccione Initial Mode y haga clic en Start.

   Cuando el estado del trabajo cambie a FINISHED, los datos ya se habrán escrito.

Paso 4: Transmitir datos desde la tabla de Paimon mediante streaming

1. Cree un borrador de flujo en blanco. Copie el siguiente código SQL en el editor SQL para enviar todos los datos de la tabla my_tbl a los registros mediante el conector print.

   CREATE TEMPORARY TABLE Print (
     dt STRING,
     id BIGINT,
     content STRING
   ) WITH (
     'connector' = 'print'
   );
   INSERT INTO Print SELECT * FROM `my-catalog`.`my_db`.`my_tbl`;

2. En la esquina superior derecha del editor SQL, haga clic en Deploy. En el cuadro de diálogo Deploy draft, configure los parámetros según las indicaciones y haga clic en Confirm.

3. En la página O&M > Deployments, haga clic en Start en la columna Actions del despliegue correspondiente. Seleccione Initial Mode y haga clic en Start.

4. En la página de detalles del despliegue, consulte los resultados del cómputo de Flink.

   1. En la página O&M > Deployments, haga clic en el nombre del despliegue correspondiente.

   2. En la pestaña Logs, dentro de Running Logs, vaya a la pestaña Running Task Managers y haga clic en un enlace de tarea en la columna Path, ID.

   3. Haga clic en Stdout para consultar los datos de Paimon consumidos.

   La salida de Stdout muestra los registros INSERT de la tabla de Paimon: +I[20240108, 1, apple], +I[20240108, 2, banana], +I[20240109, 1, cat] y +I[20240109, 2, dog]. Esto confirma que el consumo mediante streaming se completó correctamente.

Paso 5: Actualizar datos en la tabla de Paimon

1. Cree un borrador de flujo en blanco y copie el siguiente código SQL en el editor SQL.

   SET 'execution.checkpointing.interval' = '10s';
   INSERT INTO `my-catalog`.`my_db`.`my_tbl` VALUES ('20240108', 1, 'hello'), ('20240109', 2, 'world');

2. En la esquina superior derecha del editor SQL, haga clic en Deploy. En el cuadro de diálogo Deploy draft, configure los parámetros según las indicaciones y haga clic en Confirm.

3. En la página O&M > Deployments, haga clic en Start en la columna Actions del despliegue correspondiente. Seleccione Initial Mode y haga clic en Start.

   Cuando el estado del trabajo cambie a FINISHED, los datos ya se habrán actualizado.

4. En la pestaña Stdout de la página Deployments del trabajo del paso 4, consulte los datos que se actualizaron en la tabla de Paimon.

   En la pestaña Running Task Managers, haga clic en el TaskManager correspondiente y abra la subpestaña Stdout. Aparecen los registros CDC de la tabla de Paimon: -U[20240108, 1, apple], +U[20240108, 1, hello], -U[20240109, 2, dog] y +U[20240109, 2, world]. -U indica el registro antes de la actualización y +U indica el registro después de la actualización.

Paso 6 (opcional): Cancelar el trabajo y limpiar los recursos

Después de las pruebas, cancele el trabajo de consumo mediante streaming y limpie los recursos.

1. En la página O&M > Deployments, haga clic en Cancel en la columna Actions del despliegue correspondiente para detener el trabajo.

2. En el editor de la pestaña Scripts, introduzca el siguiente código para eliminar los archivos de datos de Paimon y el catálogo de Paimon.

   DROP DATABASE `my-catalog`.`my_db` CASCADE; -- Deletes all data files for the Paimon database from OSS.
   DROP CATALOG `my-catalog`; -- Removes the Paimon catalog from the Realtime Compute for Apache Flink console metadata but does not affect the data files in OSS.

   Un mensaje The following statement has been executed successfully! indica que los archivos de datos y el catálogo de Paimon se eliminaron correctamente.

Temas relacionados

• Para obtener más información sobre cómo escribir datos en una tabla de Paimon o consumirlos desde ella, consulte Write data to and consume data from Paimon tables.

• Para modificar el esquema de una tabla de Paimon (por ejemplo, agregar columnas o cambiar sus tipos) o modificar temporalmente los parámetros de la tabla, consulte Modify a table schema.

• Para conocer las optimizaciones comunes de las tablas de clave principal y las tablas Append Scalable de Paimon, consulte Paimon performance optimization.

• Para consultar las preguntas frecuentes sobre Paimon, consulte Connectors.