Un nodo Flink SQL batch permite utilizar instrucciones SQL estándar para definir y ejecutar tareas de procesamiento de datos. Úselo para analizar y transformar grandes conjuntos de datos en tareas como limpieza y agregación de datos. El nodo admite configuración visual y ofrece una solución eficiente y flexible para el procesamiento por lotes a gran escala. Este tema describe cómo usar un nodo Flink SQL batch para procesar datos en lotes.
Requisitos previos
Ha creado un espacio de trabajo y ha vinculado un recurso de cómputo de Realtime Compute for Apache Flink en Administration. Para obtener más información, consulte Bind computing resources.
Ha creado un nodo Flink SQL batch. Para obtener más información, consulte Create a node for a scheduling workflow.
-
Ha otorgado los siguientes permisos de API al usuario de RAM o rol de RAM que DataWorks utiliza para llamar a las API de Realtime Compute for Apache Flink. Estos permisos son obligatorios para enviar y desplegar tareas de nodo en un clúster de Flink. Para obtener más información, consulte Grant permissions.
{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": ["stream:CreateDeployment", "stream:UpdateDeployment", "stream:GetDeployment", "stream:DeleteDeployment"], "Resource": ["*"] } ] }
Limitación
Solo se admite un resource group serverless. El resource group exclusivo heredado para programación no es compatible.
Paso 1: Desarrollar el nodo Flink SQL batch
En la página de edición del nodo Flink SQL batch, desarrolle la tarea del nodo.
Desarrollar código SQL
Desarrolle el código de su tarea en el área de edición de SQL. En el código, puede definir variables mediante el formato ${variable_name}. Posteriormente, en el lado derecho de la página de edición del nodo, asigne un valor a la variable en la sección Scheduling Parameters del panel Scheduling Settings. Esto le permite pasar parámetros dinámicamente a su código en escenarios de programación. Para obtener más información sobre cómo usar parámetros de programación, consulte Scheduling parameter sources and their expressions. A continuación se muestra un ejemplo.
-- Create a source table named datagen_source. CREATE TEMPORARY TABLE datagen_source_${var}( name VARCHAR ) WITH ( 'connector' = 'datagen', 'number-of-rows' = '1000' ); -- Create a result table named blackhole_sink. CREATE TEMPORARY TABLE blackhole_sink_${var}( name VARCHAR ) WITH ( 'connector' = 'blackhole' ); -- Insert data from the source table into the result table. INSERT INTO blackhole_sink_${var} SELECT name FROM datagen_source_${var};
En este ejemplo, el parámetro bizdate tiene un valor de $[yyyymmdd], lo que habilita la sincronización por lotes de nuevos datos diarios.
Paso 2: Configurar el nodo Flink SQL batch
Configure los parámetros de tarea del nodo Flink SQL batch según sus requisitos empresariales.
Configurar recursos de Flink
Puede configurar los siguientes parámetros en el lado derecho de la página de edición, en la sección Flink resource information bajo Scheduling Settings. Para obtener más información, consulte Configure schedule settings.
|
Parámetro |
Descripción |
|
Flink cluster |
Nombre del recurso de cómputo de Flink totalmente gestionado vinculado en Administration. |
|
Flink engine version |
Seleccione una versión del motor según sus necesidades. |
|
Resource Group for Scheduling |
Seleccione un serverless resource group que tenga conectividad de red con Flink. |
|
Job Manager CPU |
Según las mejores prácticas de Flink, JobManager requiere como mínimo 0.5 núcleos de CPU y 2 GiB de memoria para operar de forma estable. Se recomienda 1 núcleo de CPU y 4 GiB de memoria, con un máximo de 16 núcleos de CPU. Ajuste la configuración según la escala de su clúster y la complejidad del trabajo. |
|
Job Manager Memory |
La configuración de memoria de JobManager afecta su capacidad para manejar tareas de programación y gestión. El rango recomendado es de 2 GiB a 64 GiB para garantizar un funcionamiento estable y eficiente. Ajuste el valor según la escala de su clúster y los requisitos del trabajo. |
|
Task Manager CPU |
La configuración de CPU de TaskManager influye en su capacidad de procesamiento de tareas. De acuerdo con las mejores prácticas de Flink, se recomiendan como mínimo 0.5 núcleos de CPU y 2 GiB de memoria; la configuración sugerida es de 1 núcleo de CPU y 4 GiB de memoria, hasta un máximo de 16 núcleos de CPU. Ajuste la configuración según sus necesidades. |
|
Task Manager Memory |
La configuración de memoria de TaskManager determina el volumen de datos y el rendimiento en el procesamiento de tareas. Para asegurar una ejecución estable y eficiente, el tamaño de memoria debe ser de al menos 2 GiB y puede configurarse hasta 64 GiB. |
|
Concurrency |
Este parámetro determina el número de ejecuciones paralelas de tareas en un trabajo de Flink. Una concurrencia mayor mejora la velocidad de procesamiento y la utilización de recursos. Establezca este valor según los recursos de su clúster y las características del trabajo. |
|
Maximum number of slots |
Un slot representa una unidad de recurso de tamaño fijo en un TaskManager que puede asignarse a tareas. Cada slot puede ejecutar una instancia de tarea u operador. Puede ajustar el número máximo de slots según los recursos disponibles. |
|
Number of slots per TaskManager |
El número de slots por TaskManager define cuántas tareas puede ejecutar en paralelo. Ajuste la configuración de slots para optimizar la utilización de recursos y la capacidad de procesamiento paralelo. |
(Opcional) Configurar parámetros de programación
En el lado derecho de la página de edición, dentro de la sección Scheduling Parameters bajo Scheduling Settings, haga clic en Add parameters y luego edite Parameter name y Parameter Value para utilizarlos dinámicamente en su código.
(Opcional) Configurar parámetros de ejecución de Flink
Puede configurar los parámetros de ejecución en el lado derecho de la página de edición, en la sección Flink running parameters bajo Scheduling Settings. Para obtener más información, consulte Configure schedule settings.
Al configurar parámetros de ejecución de Flink, la sintaxis es compatible con VVP (Ververica Platform). Puede escribir las configuraciones directamente en formato YAML sin agregar punto y coma ni otros caracteres especiales para los saltos de línea.
Para ejecutar la tarea del nodo en un programa periódico, configure la información de programación (Scheduling Policy, Scheduling time, Scheduling Dependency y Node output parameters) según sus requisitos empresariales. Para obtener más información, consulte Configure schedule settings.
Una vez completada la configuración de la tarea, haga clic en Save.
Paso 3: (Opcional) Depurar el nodo Flink SQL batch
Antes de desplegar el nodo en el entorno de producción, utilice la función de depuración para realizar una ejecución de prueba del código del nodo con datos simulados cargados. Esto le permite verificar la lógica SQL y los datos ascendentes y descendentes sin necesidad de desplegar la tarea en Operation Center.
La función de depuración está disponible mediante una lista de permitidos. Para usar esta función, envíe un ticket para solicitar acceso.
Configurar información de recursos de Flink
En el lado derecho de la página de edición del nodo, dentro de la sección Flink resource information del panel Run Configuration, configure los parámetros descritos en la siguiente tabla.
|
Parámetro |
Descripción |
|
Flink Debug Cluster |
Clúster de sesión de Flink utilizado para ejecutar la tarea de depuración. Este parámetro es obligatorio. La lista desplegable muestra los clústeres de sesión existentes bajo el recurso de cómputo actual junto con su estado de ejecución. Solo pueden seleccionarse los clústeres con estado Running. Si no hay clústeres disponibles en la lista, haga clic en Create Cluster para ir a la consola de Realtime Compute for Apache Flink y crear un clúster de sesión. |
|
Flink Engine Version |
Versión del motor de Flink del clúster de sesión seleccionado. El sistema muestra este valor automáticamente según el clúster y no requiere entrada manual. |
|
Timeout |
Duración máxima de una sola tarea de depuración, en minutos. El valor predeterminado es 30 minutos. La tarea de depuración se detiene automáticamente al transcurrir la duración especificada. |
Al cambiar el recurso de cómputo del nodo actual, se borran el Flink Debug Cluster seleccionado y los datos de depuración cargados. Debe volver a seleccionar un clúster y cargar nuevamente los datos.
Preparar datos de depuración
En la sección Debug Data del panel Run Configuration, prepare datos simulados para las tablas de origen referenciadas en su código.
Haga clic en Generate Template. El sistema analiza sintácticamente las tablas de origen referenciadas en el código SQL actual y genera los registros correspondientes de nombres de tabla en la lista inferior. Los datos cargados previamente no se eliminan.
En la columna Actions del registro de nombre de tabla, haga clic en Download Template para descargar una plantilla CSV que coincida con la estructura de la tabla de origen.
Complete los datos de depuración en la plantilla descargada respetando el orden de las columnas y guarde el archivo en formato CSV.
En la columna Actions del registro de nombre de tabla, haga clic en Upload y seleccione el archivo CSV completado para cargarlo. Tras una carga exitosa, la columna Status mostrará Enabled.
(Opcional) Después de cargar los datos correctamente, haga clic en Preview en el panel inferior para consultar los datos. Para modificarlos, cargue nuevamente un archivo CSV que sobrescriba los datos existentes.
Si no desea que los datos simulados de una tabla de origen específica se usen en la sesión de depuración actual, haga clic en Disable para cambiar el estado a Disabled. Para reactivar los datos, haga clic en Enable. Solo se utilizan en la sesión de depuración actual los datos con estado Enabled.
Antes de cargar datos de depuración, debe seleccionar primero un Flink Debug Cluster. De lo contrario, el sistema le pedirá que seleccione un recurso de cómputo.
Los datos de depuración solo admiten formato CSV y el tamaño del archivo no puede superar 1 MB. La primera fila del archivo CSV debe contener los nombres de las columnas. Se recomienda usar codificación UTF-8.
Ejecutar la tarea de depuración
Cuando los datos de depuración estén listos, haga clic en el botón Run de la barra de herramientas del editor (o presione F8). El sistema envía el código, los datos simulados y la información de recursos de Flink al clúster de sesión seleccionado para su ejecución.
Si su código usa parámetros en formato ${variable_name}, asegúrese de haber asignado valores a las variables en la sección Scheduling Parameters. Durante la depuración, el sistema reemplaza los marcadores de posición del código con los valores asignados antes del envío.
Consultar los resultados de depuración
Tras la ejecución de la tarea de depuración, el área de resultados en la parte inferior del nodo proporciona la siguiente información para ayudarle a identificar problemas rápidamente:
Code: Código SQL enviado al motor de Flink para esta ejecución (con las sustituciones de variables aplicadas).
Logs: Registro de ejecución e información de errores de la tarea de depuración.
Query results: Datos de salida de la tarea de depuración.
Paso 4: Desplegar y gestionar el nodo Flink SQL batch
Una vez configurada la tarea del nodo, debe desplegarlo. Para obtener más información, consulte Deploy a node.
Después de desplegar la tarea, haga clic en Go to operation and maintenance debajo de Deploy to Production para consultar el estado de ejecución de las tareas programadas en Operation Center. Para obtener más información, consulte View scheduled tasks.