擷取批次運算Metrics

0.0.201

本文由簡體中文內容自動轉碼而成。阿里雲不保證此自動轉碼的準確性、完整性及時效性。本文内容請以簡體中文版本為準。

本文主要介紹批次運算metrics統計項以及擷取方式。批次運算對外的資源主要表現在叢集和作業兩個維度，因此metrics統計也從以上2個維度展現。

叢集 Metrics

Metrics 專案

叢集對外提供的 Metrics 統計項包括如下：

統計項	名稱	單位	彙總統計方法

統計項	名稱	單位	彙總統計方法
cls_dataVfsFsSizePused	資料盤利用率	%	Average,Maximum,Minimum
cls_systemCpuLoad	CPU負載	%	Average,Maximum,Minimum
cls_systemCpuUtilIdle	CPU空閑率	%	Average,Maximum,Minimum
cls_systemCpuUtilUsed	CPU使用率	%	Average,Maximum,Minimum
cls_vfsFsSizePused	系統硬碟利用率	%	Average,Maximum,Minimum
cls_vmMemorySizePused	記憶體利用率	%	Average,Maximum,Minimum

以上統計專案以執行個體為單位進行上報，也即每個執行個體都有以上統計項；
每個metrics專案上報的資料以clusterId、GroupId、InstanceId等維度組織；
每條記錄都包括，當前統計項在過去1分鐘內的平均值、最大值以及最小值；
批次運算預設10秒鐘推送一次資料；
擷取 Metrics 時可以設定彙總周期(“Period”),預設為1分鐘。

資料記錄樣本

clsdata

控制台擷取方法

clsinfo instanceInfo instanceMetrics

作業 Metrics

Metrics 專案

作業對外提供的 Metrics 統計項包括如下：

統計項	名稱	單位	彙總統計方法

統計項	名稱	單位	彙總統計方法
job_dataVfsFsSizePused	資料盤利用率	%	Average,Maximum,Minimum
job_systemCpuLoad	CPU負載	%	Average,Maximum,Minimum
job_systemCpuUtilIdle	CPU空閑率	%	Average,Maximum,Minimum
job_systemCpuUtilUsed	CPU使用率	%	Average,Maximum,Minimum
job_vfsFsSizePused	系統硬碟利用率	%	Average,Maximum,Minimum
job_vmMemorySizePused	記憶體利用率	%	Average,Maximum,Minimum

資料記錄樣本

jobdata

控制台擷取方式

jobInfo jobMetrics

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)