Message delivery - Realtime Compute for Apache Flink - Centro de documentación de Alibaba Cloud

La entrega de mensajes transmite en tiempo real los registros de inicio, el uso de recursos, los eventos de ejecución y los cambios de estado de los trabajos de Flink en un namespace de destino hacia Simple Log Service (SLS) o ApsaraMQ for Kafka. Al conservar estos datos, usted puede consultar los registros históricos para solucionar problemas, optimizar el rendimiento y realizar auditorías. En este tema se describe cómo configurar la entrega de mensajes y consultar los mensajes entregados.

Contexto

La entrega de mensajes es compatible con SLS y ApsaraMQ for Kafka. Ambos canales transmiten el mismo contenido de mensaje. SLS admite la entrega entre regiones, mientras que Kafka solo envía mensajes a una instancia ubicada en la misma región que su espacio de trabajo de Flink. En la siguiente tabla se describen los tipos de información que se entregan y los eventos que activan dicha entrega.

Categoría	Descripción	Evento de activación
Registros de inicio del trabajo	Registros generados durante todo el proceso de inicio del trabajo, desde la inicialización del entorno de Flink hasta el arranque de JobManager y la generación del gráfico de ejecución de Flink.	Se entrega una sola vez cuando el trabajo inicia correctamente o alcanza un estado final (como failed o finished).
Información de recursos	Cambios en el uso de recursos dentro de un namespace, incluido el consumo y la asignación de CPU y memoria. Cambios en el uso de recursos de una cola, incluido el consumo y la asignación de CPU y memoria. Importante Esta información sirve únicamente para dar seguimiento a la capacidad de los recursos y no genera alertas.	Se entrega cada 30 segundos mientras el namespace opera con normalidad.
Eventos de tiempo de ejecución	El estado de inicio en cada etapa del proceso de arranque del trabajo.	Se activa cuando ocurre un evento de tiempo de ejecución del trabajo.
Consumo de recursos del trabajo	Uso de recursos de los trabajos de streaming en ejecución. No incluye trabajos por lotes ni trabajos ejecutados en un clúster de sesión.	Se entrega cada 10 minutos mientras el trabajo está en ejecución.
Cambio de estado del trabajo	Eventos de cambio de estado del trabajo, que incluyen el estado anterior y el nuevo estado.	Se activa cuando cambia el estado del trabajo.

Notas

Antes de entregar mensajes a SLS, cree un proyecto de SLS y un logstore. Para obtener instrucciones, consulte Use LoongCollector to collect and analyze text logs on ECS instances.
Antes de entregar mensajes a Kafka, cree una instancia de ApsaraMQ for Kafka y un topic en la consola de ApsaraMQ for Kafka. El topic no se crea automáticamente. Para obtener más información, consulte Create Kafka resources.
La entrega a SLS y la entrega a Kafka son independientes. Usted puede habilitar uno o ambos canales y configurar el rango de entrega por separado en cada pestaña.
La entrega de mensajes no tiene costo. Sin embargo, el uso de funciones de SLS, como la indexación del logstore, sí genera cargos. Para obtener más información, consulte Billing overview.
Consultar y analizar registros en SLS requiere un índice. Habilitar un índice es opcional y genera cargos por tráfico y almacenamiento de índices. Para conocer más sobre estos cargos, consulte Billing overview.
Es posible configurar el cifrado del lado del servidor para el logstore. Los mensajes que se entreguen al logstore también se cifrarán. Para saber cómo hacerlo, consulte Data encryption.
La entrega de mensajes abarca los registros de inicio, los detalles de recursos, todos los eventos de ejecución, el uso de recursos del trabajo y los cambios de estado del trabajo. Para enviar los registros de tiempo de ejecución del trabajo, consulte Configure job log output.
Los cambios en la configuración de entrega de mensajes pueden tardar hasta 30 segundos en surtir efecto.

Procedimiento

Paso 1: Configurar la entrega de mensajes

Vaya a la página de configuración de entrega de mensajes
1. Inicie sesión en la consola de Realtime Compute for Apache Flink.
2. En la columna Actions del espacio de trabajo de destino, haga clic en Console.
3. En el panel de navegación izquierdo, haga clic en O&M > Configurations.
4. Haga clic en la pestaña Message Delivery.

Configure los parámetros de entrega de mensajes

Complete la configuración en la pestaña del canal de destino.

SLS Delivery

Active el interruptor Deliver to SLS.

Configure los parámetros de Log Service.

Parámetro	Descripción
Authorization mode	STS Token: Este modo entrega mensajes únicamente a un proyecto de Log Service ubicado en la misma región que su espacio de trabajo de Flink. Por lo tanto, solo necesita configurar el proyecto de Log Service y el logstore. AccessKey: Este modo permite entregar mensajes a un proyecto de Log Service en una región distinta. Por ello, debe configurar la región, el endpoint, el AccessKey ID y el AccessKey Secret.
Log Service project	El nombre del proyecto de Log Service.
Log Service Logstore	El nombre del logstore de Log Service.
Region	La región donde reside el proyecto de Log Service. Si Authorization mode se establece en AccessKey, debe configurar este parámetro. Para conocer las regiones y los endpoints, consulte Endpoints.
endpoint	El endpoint de Log Service. Si Authorization mode se establece en STS Token, el sistema utiliza automáticamente el endpoint de la región donde se encuentra su espacio de trabajo de Flink. Si configura Authorization mode en AccessKey, debe establecer este parámetro manualmente.
Delivery range	El contenido del mensaje que se va a entregar. Para obtener más información, consulte Field reference más adelante en este tema.
AccessKey ID	El AccessKey ID y el AccessKey Secret de su cuenta de Alibaba Cloud. Importante Para evitar que sus credenciales de AccessKey queden expuestas, se recomienda gestionar el AccessKey ID y el AccessKey Secret como variables. Haga clic en la flecha desplegable para seleccionar una variable existente o haga clic en el ícono junto al parámetro para crear una nueva variable. Para obtener más información sobre cómo gestionar variables y consultar su AccessKey ID y AccessKey Secret, consulte Variable management y How do I view AccessKey credentials?
AccessKey Secret

Haga clic en Save.

Kafka Delivery

Active el interruptor Deliver to Kafka.

Configure los parámetros de Kafka.

Parámetro	Descripción
Authorization Mode	STS Token: Entrega mensajes únicamente a una instancia de Kafka ubicada en la misma región que su espacio de trabajo de Flink. Los campos Kafka instance, Endpoint y Kafka Topic se seleccionan desde listas desplegables. Manual configuration: Introduzca Endpoint y Kafka Topic de forma manual.
Kafka instance	La instancia de Kafka que recibe los mensajes. Se muestra cuando el modo de autorización es STS Token. Haga clic en el ícono de actualización para volver a cargar la lista de instancias.
Endpoint	El endpoint de la instancia de Kafka. Se completa automáticamente según la instancia seleccionada cuando el modo de autorización es STS Token. Debe introducirse manualmente si el modo de autorización es Manual configuration.
Kafka Topic	El topic que recibe los mensajes. Cree el topic con anticipación. Haga clic en Kafka console para ir a la consola de ApsaraMQ for Kafka. Cuando el modo de autorización es Manual configuration, el nombre del topic debe tener entre 3 y 64 caracteres. Solo puede contener letras, dígitos, puntos (.), guiones (-) y guiones bajos (_), e incluir como mínimo una letra o un dígito.
Delivery range	Los tipos de mensajes que se van a entregar. Seleccione como mínimo una de las siguientes opciones: Startup Logs, Resource Details, Full Running Events, Job Resource Usage o Job Status Changed. Para obtener más información sobre el contenido de los mensajes, consulte Field descriptions más adelante en este tema.

Haga clic en Save.

Paso 2: Consultar los mensajes entregados

Procedimiento

Deliver to SLS

Haga clic en Open the SLS console, ubicado a la derecha de Log Service project.
Consulte la información en los registros sin procesar.

Los registros sin procesar son mensajes en formato JSON que contienen información sobre la cuota de recursos (RESOURCE_QUOTA). Estos registros incluyen campos como namespaceTotalCpu (20.0), namespaceTotalMemory (0.08 TiB), namespaceUsedCpu (8.0), namespaceUsedMemory (23.00 GiB) y resourceQueueName (default-queue). Dichos campos registran las cuotas y el uso de CPU y memoria del namespace y de la cola de recursos.

Deliver to Kafka

Los mensajes se entregan en formato JSON al topic de Kafka que configuró en el paso 1.
Inicie sesión en la consola de ApsaraMQ for Kafka y consulte los mensajes en la página Message query de la instancia de destino. Para obtener más información, consulte Query messages. También puede usar un cliente de Kafka autogestionado para consumir el topic a través del Endpoint configurado.

Descripción de los campos

La estructura del cuerpo del mensaje es idéntica para SLS y Kafka. A continuación se describen los campos de cada tipo de mensaje.

Startup logs (JOB_START_LOG)

Parámetro	Descripción
messageType	Tipo de mensaje. Valor fijo: JOB_START_LOG (registro de inicio del trabajo).
deploymentId	El ID del despliegue del trabajo.
deploymentName	El nombre del despliegue del trabajo.
jobId	El ID de la instancia del trabajo.
tag	La etiqueta del trabajo. Este campo queda vacío si no se configuró ninguna etiqueta.
length	La longitud total del registro.
offset	Para registros fragmentados, este campo especifica el desplazamiento inicial de la entrada del registro.
content	Detalles del registro de inicio del trabajo.
workspace	El ID del espacio de trabajo.
namespace	El nombre del namespace.
messageId	El ID único del mensaje.
timestamp	La marca de tiempo en la que se generó el mensaje.

Resource details (JOB_RESOURCE_QUOTA)

Parámetro	Descripción
messageType	Tipo de mensaje. Valor fijo: RESOURCE_QUOTA (uso de recursos).
namespaceTotalCpuMemory	Este campo está obsoleto. Utilice en su lugar los campos correspondientes `...Cpu` y `...Memory`.
namespaceTotalCpu	El total de recursos de CPU del namespace, medido en CU.
namespaceTotalMemory	El total de recursos de memoria del namespace.
namespaceUsedCpuMemory	Este campo está obsoleto. Utilice en su lugar los campos correspondientes `...Cpu` y `...Memory`.
namespaceUsedCpu	Los recursos de CPU que utiliza el namespace, medidos en CU.
namespaceUsedMemory	Los recursos de memoria que utiliza el namespace.
resourceQueueName	El nombre de la cola de recursos.
resourceQueueTotalCpuMemory	Este campo está obsoleto. Utilice en su lugar los campos correspondientes `...Cpu` y `...Memory`.
resourceQueueTotalCpu	El total de recursos de CPU de la cola de recursos, medido en CU.
resourceQueueTotalMemory	El total de recursos de memoria de la cola de recursos.
resourceQueueUsedCpuMemory	Este campo está obsoleto. Utilice en su lugar los campos correspondientes `...Cpu` y `...Memory`.
resourceQueueUsedCpu	Los recursos de CPU que utiliza la cola de recursos, medidos en CU.
resourceQueueUsedMemory	Los recursos de memoria que utiliza la cola de recursos.
workspace	El ID del espacio de trabajo.
namespace	El nombre del namespace.
messageId	El ID único del mensaje.
timestamp	La marca de tiempo en la que se generó el mensaje.

Runtime events (JOB_EVENT)

Parámetro	Descripción
messageType	Tipo de mensaje. Valor fijo: JOB_EVENT (evento de tiempo de ejecución).
deploymentId	El ID del despliegue del trabajo.
deploymentName	El nombre del despliegue del trabajo.
jobId	El ID de la instancia del trabajo.
tag	La etiqueta del trabajo. Este campo queda vacío si no se configuró ninguna etiqueta.
eventId	El ID del evento.
eventName	El nombre del evento.
content	Detalles del evento.
workspace	El ID del espacio de trabajo.
namespace	El nombre del namespace.
messageId	El ID único del mensaje.
timestamp	La marca de tiempo en la que se generó el mensaje.

Job resource usage (JOB_RESOURCE_USAGE)

Parámetro	Descripción
messageType	Tipo de mensaje. Valor fijo: JOB_RESOURCE_USAGE (uso de recursos del trabajo).
deploymentId	El ID del despliegue del trabajo.
deploymentName	El nombre del despliegue del trabajo.
jobId	El ID de la instancia del trabajo.
tag	La etiqueta del trabajo. Este campo queda vacío si no se configuró ninguna etiqueta.
jobUsedCpu	La cantidad de CU que utiliza el trabajo.
jobUsedMemory	La cantidad de memoria que utiliza el trabajo.
workspace	El ID del espacio de trabajo.
namespace	El nombre del namespace.
messageId	El ID único del mensaje.
timestamp	La marca de tiempo en la que se generó el mensaje.

Job status changes (JOB_STATUS_CHANGED)

Parámetro	Descripción
messageType	Tipo de mensaje. Valor fijo: JOB_STATUS_CHANGED (cambio de estado del trabajo).
messageId	El ID único del mensaje.
timestamp	La marca de tiempo en la que se generó el mensaje, en milisegundos.
workspace	El ID del espacio de trabajo.
namespace	El nombre del namespace.
deploymentId	El ID del despliegue del trabajo.
deploymentName	El nombre del despliegue del trabajo.
jobId	El ID de la instancia del trabajo.
previousStatus	El estado del trabajo antes del cambio.
currentStatus	El estado del trabajo después del cambio.
tag	La etiqueta del trabajo. Este campo queda vacío si no se configuró ninguna etiqueta.
additionalInfo	Información adicional.

Referencias

Para configurar el registro de un solo trabajo, consulte Configure job log output.
Para consultar registros en la consola de desarrollo de Realtime Compute for Apache Flink, consulte View startup and running logs, View running events, View running exception logs y View historical job instance logs.