Realtime Compute for Apache Flink:Trabajo de Flink SQL

Requisitos previos

• El usuario de RAM o el rol de RAM que utiliza cuenta con los permisos obligatorios para la consola de Realtime Compute. Para obtener más información, consulte permissions.

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

Paso 1: Crear un borrador de SQL

1. Vaya a la página de creación de borradores de SQL.

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

   2. Localice el espacio de trabajo de Flink deseado y haga clic en Console en la columna Actions.

   3. En el panel de navegación, haga clic en Development > ETL.

2. Haga clic en el icono y luego en New Blank Stream Draft. Introduzca un Name y seleccione una Engine Version.

   Realtime Compute for Apache Flink ofrece diversas plantillas de código y de sincronización de datos. Cada plantilla incluye descripciones de casos de uso, ejemplos de código e instrucciones. Puede hacer clic en una plantilla para familiarizarse rápidamente con las funciones del producto y la sintaxis necesaria para implementar su lógica de negocio. Para obtener más información, consulte Code templates y Data synchronization templates.

   Parámetro	Descripción	Ejemplo
   Name	Nombre del borrador de SQL. Nota El nombre debe ser único dentro del proyecto actual.	flink-test
   Engine Version	Versión del motor de Flink para el borrador de SQL. Se recomienda utilizar versiones con la etiqueta RECOMMENDED o Stable para garantizar mayor confiabilidad y rendimiento. Para obtener más información sobre las versiones del motor, consulte Release notes y Engine versions.	vvr-8.0.8-flink-1.17

3. Haga clic en Create.

Paso 2: Escribir SQL y consultar la configuración del borrador

1. Escriba el código SQL.

   Copie el siguiente código SQL en el editor. Este ejemplo utiliza un conector Datagen para generar un flujo de datos aleatorio y un conector Print para escribir la salida en los registros de la consola. Para obtener más información sobre los conectores compatibles, consulte Supported connectors.

   -- Create a temporary source table named datagen_source.
   CREATE TEMPORARY TABLE datagen_source(
     randstr VARCHAR
   ) WITH (
     'connector' = 'datagen' -- Use the Datagen connector.
   );
   
   -- Create a temporary sink table named print_table.
   CREATE TEMPORARY TABLE print_table(
     randstr  VARCHAR
   ) WITH (
     'connector' = 'print',   -- Use the Print connector.
     'logger' = 'true'        -- Write the output to logs.
   );
   
   -- Select a substring from the randstr field and insert it into the sink table.
   INSERT INTO print_table
   SELECT SUBSTRING(randstr,0,8) from datagen_source;

   Nota

   • Este ejemplo emplea una instrucción INSERT INTO para escribir datos en una sola tabla de destino. También es posible usar la instrucción INSERT INTO para escribir datos en varias tablas de destino. Para obtener más información, consulte INSERT INTO statement.

   • En entornos de producción, utilice tablas registradas en catálogos en lugar de tablas temporales. Para obtener más información, consulte Catalogs.

2. Consulte la configuración del borrador.

   A la derecha del editor SQL, puede consultar o configurar ajustes en varias pestañas.

   Pestaña	Descripción
   Configurations	Engine version: Para obtener más información, consulte Engine versions y Lifecycle policies. Se recomienda usar una versión recomendada o estable. Las etiquetas de versión son las siguientes: Recommend: La última versión menor de la versión principal actual. Stable: La última versión menor de una versión principal dentro de su periodo de soporte, con defectos conocidos corregidos. Normal: Otras versiones menores que aún se encuentran dentro de su periodo de soporte. Deprecated: Versiones que han superado su fecha de fin de vida útil (EOL). Additional dependencies: Dependencias adicionales necesarias para el trabajo, como funciones temporales. Kerberos Authentication: Habilite la autenticación Kerberos y configure un clúster Kerberos registrado junto con la información del principal. Si no ha registrado un clúster Kerberos, consulte Register a Hive Kerberos cluster.
   Structure	Flow Diagram: Utilice el diagrama de flujo de datos para visualizar rápidamente el linaje de los datos. Tree Diagram: Utilice el diagrama de árbol para identificar rápidamente los orígenes de datos.
   Versions	Aquí puede consultar el historial de versiones del borrador de SQL. Para obtener más información sobre las funciones disponibles en la columna Actions, consulte Manage draft versions.

(Opcional) Paso 3: Validar y depurar el borrador de SQL

1. Valide el borrador de SQL.

   La validación comprueba la semántica SQL, la conectividad de red y los metadatos de las tablas utilizadas en el borrador. Una vez completada la validación, puede hacer clic en SQL Advice en el área de resultados para consultar posibles riesgos y sugerencias de optimización de SQL.

   1. En la esquina superior derecha del editor SQL, haga clic en Validate.

   2. En el cuadro de diálogo Validate, haga clic en Confirm.

   Nota

   Si se produce un error de tiempo de espera, es posible que aparezca el siguiente mensaje: The RPC times out maybe because the SQL parsing is too complicated. Please consider enlarging the flink.sqlserver.rpc.execution.timeout option in flink-configuration, which by default is 120 s.

   Solución: Agregue el siguiente parámetro de configuración al principio del editor SQL.

   SET 'flink.sqlserver.rpc.execution.timeout' = '600s';

2. Depure el borrador de SQL.

   La función de depuración permite simular la ejecución de un trabajo para revisar su salida y verificar la lógica SELECT o INSERT. Esta función mejora la eficiencia del desarrollo y reduce los riesgos relacionados con la calidad de los datos.

   Nota

   La función de depuración no escribe datos en la tabla de destino.

   1. En la esquina superior derecha del editor SQL, haga clic en Debug.

   2. En el cuadro de diálogo Debug, seleccione un clúster de depuración y haga clic en Next.

      Si no hay ningún clúster de sesión disponible, deberá crear uno. El clúster de sesión debe utilizar la misma versión del motor que el borrador de SQL y encontrarse en estado Running. Para obtener más información, consulte Step 1: Create a session cluster.

   3. Configure los datos de depuración y haga clic en OK.

      Para obtener más información sobre la configuración, consulte Step 2: Debug a job.

Paso 4: Desplegar el borrador de SQL

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 sea necesario y haga clic en OK.

Al desplegar el borrador, puede seleccionar una Resource Queue o un session cluster como Deployment target. La siguiente tabla compara ambas opciones.

Paso 5: Iniciar el trabajo y consultar los resultados

1. En el panel de navegación, haga clic en O&M > Deployments.

2. Localice el trabajo deseado y haga clic en Start en la columna Actions.

   Seleccione Initial Mode y luego haga clic en Start. El trabajo se encuentra en ejecución cuando su estado cambia a Running. Para obtener más información sobre los parámetros de inicio, consulte Start a job.

3. En la página de detalles de Deployments, consulte los resultados del trabajo de Flink.

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

   2. En la pestaña Logs, haga clic en la subpestaña Running Task Managers. En la columna Path, ID, haga clic en un TaskManager.

   3. Haga clic en la pestaña Logs y busque registros relacionados con PrintSinkOutputWriter.

      Si encuentra entradas de registro que contienen la cadena Source: datagen_source → Calc → Sink: print_table, significa que el flujo de datos se está procesando correctamente.

(Opcional) Paso 6: Detener el trabajo

Para aplicar cambios a un trabajo (como modificaciones de código, actualizaciones de parámetros WITH o cambios de versión), debe volver a desplegarlo, detenerlo y reiniciarlo. También es necesario reiniciar para realizar un inicio sin estado o aplicar cambios de configuración no dinámicos. Para obtener más información sobre cómo detener un trabajo, consulte Stop a job.

1. En la página O&M > Deployments, localice el trabajo deseado y haga clic en Cancel en la columna Actions.

2. Haga clic en OK.

Documentos relacionados

• Preguntas frecuentes sobre desarrollo y O&M de trabajos

  • Job development

  • Job O&M

• Configurar información del trabajo

  • Puede configurar los recursos antes de iniciar un trabajo o modificarlos para un despliegue en ejecución. Se admiten dos modos de configuración de recursos: básico (grano grueso) y experto (grano fino). Para obtener más información, consulte Configure job resources.

  • Puede configurar niveles de registro y especificar diferentes salidas para cada nivel. Para obtener más información, consulte Configure job log outputs.

• Flujos de trabajo de desarrollo para otros tipos de trabajos

  • Flink JAR job

  • Flink Python job

  • Real-time database ingestion

• Mejores prácticas para Flink

  • Build a real-time data warehouse with Hologres

  • Build a streaming lakehouse with Paimon and StarRocks