消息投递指南-实时计算 Flink版(Flink)-阿里云帮助中心

文档备案控制台

通过消息投递功能，您可以将目标项目空间下Flink作业的启动日志、资源使用情况、运行事件及作业状态变更消息实时传输至指定的外部消息队列或存储系统。该功能可以满足数据持久化、历史记录保存、灵活集成和实时分析的需求，同时便于随时查询历史数据，为问题排查、性能优化及审计分析提供支持。本文为您介绍消息投递的操作与查看方法。

背景信息

支持跨地域投递，目前支持投递信息的类型及投递时机详情如下。

类别	投递说明	投递时机
作业启动日志	投递作业启动后的日志，涵盖从Flink环境被初始化到Job Manager启动，并生成Flink运行图的整个启动过程所产生的日志。	作业启动成功或者作业达到终态（失败或结束），投递一次启动日志。
资源情况	投递项目空间内资源的使用变化情况：包括CPU、内存资源的消耗和分配的变化情况。投递队列资源的使用变化情况：包括CPU、内存资源的消耗和分配的变化情况。重要仅用于资源容量管理的追踪，不具备告警功能。	项目空间正常运行情况下，每30秒发送一次资源情况。
运行事件	投递作业启动的过程中每个时间点的启动状态。	作业运行事件生成，触发投递。
作业消耗资源	获取运行中流作业的资源使用情况。其中不包括批作业和Session集群运行的作业。	作业运行情况下，每十分钟发送一次资源消耗情况。
作业状态变更	投递作业状态变更事件，包含状态变化前后信息。	作业状态实际发生变更时触发投递。

注意事项

目前仅支持将消息投递至SLS，因此您需要创建日志服务Project和Logstore，操作详情请参见使用LoongCollector采集并分析ECS文本日志。
投递功能本身不收费，SLS的使用（日志库索引等）会产生流量自费，详情请参见计费概述。
SLS中查询、分析日志需要开启索引，开启后会产生索引流量和索引存储空间，费用说明请参见计费概述，是否创建索引由您自主决定。
您可设置日志库的服务器端加密方式，会话记录投递会继承该加密方式，具体操作请参见数据加密。
消息投递功能目前仅支持将作业启动日志投递到SLS，作业运行日志可通过其他方式输出至OSS、SLS或Kafka，具体配置方法请参见配置作业日志输出。
变更消息投递配置，会有最长30s的延迟。

操作步骤

步骤一：配置消息投递渠道

进入消息投递配置管理入口
1. 登录实时计算管理控制台。
2. 单击目标工作空间操作列下的控制台。
3. 在左侧导航栏上，单击运维中心 > 配置管理。

配置SLS消息投递参数

在消息投递配置页签，打开投递到日志服务SLS开关。

配置SLS相关信息。

配置项	说明
授权模式	STS Token：选择此模式，只能投递到Flink工作空间所在地域的SLS项目下的日志库。因此您仅需要再配置日志服务的项目和日志库。 AccessKey：选择此模式，支持投递到其他地域的SLS项目下的日志库。因此您需要再配置endpiont、AccessKeyId和AccessKeySecret。
日志服务项目	SLS项目名称。
日志服务日志库	SLS LogStore。
Endpoint	SLS服务的Endpoint地址。当授权模式为STS Token时，系统会默认设置为Flink工作空间所在地域对应的endpoint；当授权模式为AccessKey，需要您手动配置.
投递范围	具体消息内容，详情请参见下文字段含义说明。
AccessKeyId	阿里云账号的AccessKey ID和AccessKey Secret。重要为了避免AK信息泄露，此处通过变量的方式来管理您的AK信息。您可以单击下拉箭头选择已创建的变量名，或单击参数右侧的直接创建新变量。有关变量管理及如何查看AccessKey ID和AccessKey Secret值的详细信息，请参见变量管理和如何查看AccessKey ID和AccessKey Secret信息？
AccessKeySecret

单击保存。

步骤二：查看投递消息

操作步骤

单击日志服务项目右侧的打开SLS控制台。
查看原始日志的信息。

原始日志为JSON格式的资源配额（RESOURCE_QUOTA）信息，包含namespaceTotalCpu（20.0）、namespaceTotalMemory（0.08 TiB）、namespaceUsedCpu（8.0）、namespaceUsedMemory（23.00 GiB）、resourceQueueName（default-queue）等字段，记录命名空间及资源队列的CPU与内存配额及使用量。

字段含义说明

启动日志（JOB_START_LOG）

字段	含义
messageType	消息类型，取值固定为JOB_START_LOG（作业启动日志）。
deploymentId	已部署作业ID。
deploymentName	已部署作业名称。
jobId	作业实例ID。
tag	作业标签（如无配置为空）。
length	日志总长度。
offset	用于日志分片时标记该条日志的起始位置。
content	作业启动日志详情。
workspace	工作空间ID。
namespace	项目空间名称。
messageId	消息ID。
timestamp	时间戳。

资源详情（JOB_RESOURCE_QUOTA）

字段	含义
messageType	消息类型，取值固定为RESOURCE_QUOTA（资源使用）。
namespaceTotalCpuMemory	项目空间总共的资源CU数。
namespaceTotalCpu	项目空间总共的资源CU数。
namespaceTotalMemory	项目空间总共的内存资源量，
namespaceUsedCpuMemory	项目空间已使用的资源CU数。
namespaceUsedCpu	项目空间已使用的资源CU数。
namespaceUsedMemory	项目空间已使用的内存资源量。
resourceQueueName	资源队列名称。
resourceQueueTotalCpuMemory	资源队列总资源CU数。
resourceQueueTotalCpu	资源队列总资源CU数。
resourceQueueTotalMemory	资源队列总内存资源量。
resourceQueueUsedCpuMemory	资源队列已使用的资源CU数。
resourceQueueUsedCpu	资源队列已使用的资源CU数。
resourceQueueUsedMemory	资源队列已使用的内存资源量。
workspace	工作空间ID。
namespace	项目空间名称。
messageId	消息ID。
timestamp	时间戳。

运行事件全量投递（JOB_EVENT）

字段	含义
messageType	消息类型，取值固定为JOB_EVENT（运行事件）。
deploymentId	已部署作业ID。
deploymentName	已部署作业名称。
jobId	作业实例ID。
tag	作业标签（如无配置为空）
eventId	事件ID。
eventName	事件名称。
content	作业启动日志详情。
workspace	工作空间ID。
namespace	项目空间名称。
messageId	消息ID。
timestamp	时间戳。

作业消耗资源（JOB_RESOURCE_USAGE）

字段	含义
messageType	消息类型，取值固定为JOB_RESOURCE_USAGE（作业消耗资源）。
deploymentId	已部署作业ID。
deploymentName	已部署作业名称。
jobId	作业实例ID。
tag	作业标签（如无配置为空）。
jobUsedCpu	作业已使用CU数。
jobUsedMemory	作业已使用内存数。
workspace	工作空间ID。
namespace	项目空间名称。
messageId	消息ID。
timestamp	时间戳。

作业状态变更（JOB_STATUS_CHANGED）

字段	含义
messageType	消息类型，取值固定为JOB_STATUS_CHANGED（作业状态变更）。
messageId	消息唯一标识。
timestamp	消息产生的时间戳（毫秒）。
workspace	工作空间名称。
namespace	项目空间名称。
deploymentId	已部署作业ID。
deploymentName	已部署作业名称。
jobId	作业实例ID。
previousStatus	变更前的作业状态。
currentStatus	变更后的作业状态。
tag	作业标签（如无配置为空）。
additionalInfo	附加信息。

相关文档

单作业的日志配置方法，请参见配置作业日志输出。
实时计算开发控制台上查看日志的操作方法详情，请参见查看启动和运行日志、查看运行事件、查看运行异常日志和查看历史作业实例日志。

上一篇：查看Flink审计事件下一篇：作业状态管理

该文章对您有帮助吗？