获取批量计算Metrics

本文主要介绍批量计算metrics统计项以及获取方式。批量计算对外的资源主要表现在集群和作业两个维度，因此metrics统计也从以上2个维度展现。

集群 Metrics

Metrics 项目

集群对外提供的 Metrics 统计项包括如下：

• 以上统计项目以实例为单位进行上报，也即每个实例都有以上统计项；

• 每个metrics项目上报的数据以clusterId、GroupId、InstanceId等维度组织；

• 每条记录都包括，当前统计项在过去1分钟内的平均值、最大值以及最小值；

• 批量计算默认10秒钟推送一次数据；

• 获取 Metrics 时可以设置聚合周期(“Period”),默认为1分钟。

数据记录示例

控制台获取方法

作业 Metrics

Metrics 项目

作业对外提供的 Metrics 统计项包括如下：

数据记录示例

控制台获取方式

相关 API

批量计算所有的Metrics统计信息都推送到阿里云云监控服务中。所有 Metrics 获取依赖云监控API。

DescribeMetricMetaList

查询 Metrics 项，通过该接口可以获取批量计算对外提供的统计项。

DescribeMetricData

查询 Metrics 统计数据，通过该接口可以获取指定的集群或者作业的各个统计项的数据

Demo 示例代码

#!/usr/bin/env python
#coding=utf-8
# https://help.aliyun.com/document_detail/51936.html?spm=a2c4g.11186623.6.692.347048d34VV7RU
import os
import json
import time
import sys
import datetime
from functools import wraps
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcms.request.v20190101.DescribeMetricListRequest import DescribeMetricListRequest
from aliyunsdkcms.request.v20190101.DescribeMetricMetaListRequest import DescribeMetricMetaListRequest
akId = 'AccessKeyId'
akKey = 'AccessKeySecret'
region = 'cn-hangzhou'
# jobId = "job-000000005D16F74B00006883000303E9"
jobId = "job-000000006124E89100002B3D028E3322"
def retryWrapper(func):
    @wraps(func)
    def wrapper(*args,**kwargs):
        index = 0
        while True:
            try:
                res = func(*args,**kwargs)
                break
            except Exception as e:
                if index > 6:
                    raise Exception(str(e))
                else:
                    time.sleep(0.5 * pow(2,index))
                    index += 1
        return res
    return wrapper
@retryWrapper
def listBatchMetricMeta(client, objId):
    metrics = []
    request = DescribeMetricMetaListRequest()
    request.set_accept_format('json')
    request.set_Namespace("acs_batchcomputenew")
    response = client.do_action_with_exception(request)
    res = json.loads(response)
    prefix = objId.strip().split("-")[0]
    for metric in res["Resources"]["Resource"]:
        if prefix not in metric["MetricName"]:
            continue
        metrics.append(metric["MetricName"])
    return metrics
@retryWrapper
def getSpecJobMetricsInfo(client, objId, metrics, startTime = None):
    nextToken = None
    request = DescribeMetricListRequest()
    request.set_accept_format('json')
    request.set_Period("60")
    request.set_Length("1000")
    request.set_EndTime(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(time.time())))
    # 默认查询7天的记录
    if not startTime:
        sevenDayAgo = (datetime.datetime.now() - datetime.timedelta(days = 7))
        startTime = sevenDayAgo.strftime("%Y-%m-%d %H:%M:%S")
    request.set_StartTime(startTime)
    prefix = objId.strip().split("-")[0]
    if "job" in prefix:
        dimensionInfo = [{"jobId":objId}]
    else:
        dimensionInfo = [{"clusterId":objId}]
    request.set_Dimensions(json.dumps(dimensionInfo))
    request.set_MetricName(metrics)
    request.set_Namespace("acs_batchcomputenew")
    metricsInfo = []
    while True:
        if nextToken:
            request.set_NextToken(nextToken)
        response = client.do_action_with_exception(request)
        res = json.loads(response)
        if "Datapoints" in res and len(res["Datapoints"]):
            metricsInfo.extend(json.loads(res["Datapoints"]))
        else:
            print(res)
        if "NextToken" in res and res["NextToken"]:
            nextToken = res["NextToken"]
            continue
        else:
            break
    return metricsInfo
if __name__ == "__main__":
    client = AcsClient(akId, akKey, region)
    # metricsName = ['job_systemCpuUtilIdle', 'job_systemCpuLoad', 'job_vmMemorySizePused', 'job_vfsFsSizePused', 'job_dataVfsFsSizePused']
    metricsName = listBatchMetricMeta(client, jobId)
    for metrics in metricsName:
        try:
            ret = getSpecJobMetricsInfo(client, jobId, metrics)
        except Exception as e:
            print("get metrics info failed, %s" % str(e))
            sys.exit(1)
        if not len(ret):
            continue
        # 可以对返回的数据进行二次聚合
        print(ret)

• 执行示例之前，安装阿里云 SDK 库：

• pip install aliyun_python_sdk_cms

• pip install aliyun_python_sdk_core

• 设置的 AK 账号必须要有 ‘AliyunCloudMonitorReadOnlyAccess’ 权限，添加权限的方式参考开通批量计算服务 5.2 章节；

OpenAPI 获取方式

通过 OpenAPI 可以更快的更简单的获取数据信息，只需要输入基本信息自动生成相关脚本。