利用MaxCompute InformationSchema与阿里云交易和账单管理API 实现MaxCompute费用对账分摊统计
一、需求场景分析
很多的企业用户选择MaxCompute按量付费模式构建自己的数据平台，利用MaxCompute按作业付费的计费模型，在获得高性能的同时避免"IDLE"状态的不必要资源费用支出，仅为实际使用付费。
那么在一个规模比较大的公司，企业购买了MaxCompute服务，会支撑企业内部的不同部门、个人来使用MaxCompute来开展数据处理分析。为了更好地识别数据平台使用方的周期性花费成本，优化数据资源的使用，就有必要对作业的费用进行统计，从而确认不同人员或归属部门的作业数量、作业费用、作业时长、作业资源使用量等指标。基于这些指标进行成本分摊、作业优化等管理工作。
阿里云交易和账单系统包含了MaxCompute产品的费用信息及费用明细，通过关联交易和账单系统的计费明细与MaxCompute项目的作业明细或某时间段的账单费用，可以获得使用用户、作业明细信息（如提交人账号、作业SQL内容、作业资源使用大小等信息）与计费明细或账单费用间的关系，从而开展分析。
本文将介绍如果自动化实现MaxCompute按量付费项目的作业费用统计，您也可以通过阿里云交易和账单系统API获取其他需要的费用信息，扩展分析场景。
二、方案设计思路
1、获得MaxCompute项目历史作业明细
MaxCompute Information_Schema服务是MaxCompute产品的开放元数据服务，通过Information_Schema提供的一组视图，用户可以自助地查询访问到项目内的准实时的table,column,function等全量元数据信息，同时也提供了项目内近期的作业历史明细，供使用者自助查询使用。
通过元数据服务Information_Schema里面的作业历史表tasks_history视图，可以查询到准实时的项目作业历史明细。包括：项目名称、任务名称、Instance id、开始时间、结束时间、任务复杂度、任务CPU使用情况等字段。

备注：Information_Schema目前正在灰度中，即将全面开放。
2、获取作业的计费明细数据
用户可以通过费用中心账号总览消费记录去查询具体的消费情况。
同时，格式阿里云交易和账单管理OpenAPI为用户提供管理阿里云产品售卖和财资能力，通过该api可以程序化获取MaxCompute作业计费明细数据。
调用QueryUserOmsData接口（阿里云的账单系统OMS），可以查询到具体计量信息编号、数据分类、存储、SQL读取量、公网上下行流量等字段信息。
3、关联计费明细与作业明细
通过表关联，查询到需要计算的数据结果

select
distinct
t.task_schema,
o.MeteringId,
t.owner_id,
t.operation_text,
o.type,
o.endtime,
o.computationsqlinput,
o.computationsqlcomplexity,
t.cost_cpu,o.starttime,
t.cost_mem
from  information_schema.tasks_history t
right join OdpsFeeDemo o
on t.inst_id = o. meteringid
and t.task_schema = o.projectid
where o.type = "ComputationSql";

这些数据可以通过作业ID与计费明细数据进行关联后，您就获取各个作业明细的费用信息(例如,SQL费用=数据扫描量*复杂度) ，从而可以开展不同视角的分析。

需要强调的是：MaxCompute的计费都是以阿里云费用中心的出账结果及费用明细为准。

三、具体实现步骤(含参考代码)
1.查询元数据服务里面的作业历史表tasks_history
例如，您登录访问的当前项目为 myproject1，在 myproject1 中，可以通过查询 INFORMATION_SCHEMA.tables 获得当前 myproject1 中所有表的元数据信息。

odps@ myproject1 > select * from information_schema.tables;

INFORMATION_SCHEMA 同时包含了作业历史视图，可以查询到当前项目内的作业历史信息，使用时注意添加日期分区进行过滤，例如。

odps@ mypoject1 > select * from information_schema.tasks_history where ds=’yyyymmdd’ limit 100;
odps@ myproject1 > desc package information_schema.systables;

查询历史表字段属性

odps@ myproject1 > desc information_schema.tasks_history;

如下如所示：

2.使用阿里云交易和账单管理API获取费用明细和分摊统计
方法1：手工下载上传方式
（一）首先在MaxCompute中创建结果输出表OMS表，建表语句如下：

CREATE TABLE IF NOT EXISTS OdpsFeeDemo(
ProjectId          STRING COMMENT '项目编号',
MeteringId         STRING COMMENT '计量信息编号',
Type                  STRING COMMENT '数据分类',
Storage            STRING COMMENT '存储(Byte)',
EndTime            STRING COMMENT '结束时间',
ComputationSqlInput  STRING COMMENT 'SQL读取量',
ComputationSqlComplexity  STRING COMMENT 'SQL复杂度',
StartTime          STRING COMMENT '开始时间',
OdpsSpecCode       STRING COMMENT '规格类型'
);

方法一：手动从视图下载oms账单详细费用，将数据上传（tunnel upload）到odps对应输出表

手动下载步骤：
https://help.aliyun.com/product/87964.html?spm=a2c4g.750001.list.245.5e907b138Ik9xM

进入阿里云用户中心：https://usercenter2.aliyun.com/home
返回旧版

费用中心>消费记录>使用记录
选择产品类型，填写使用期间，计算粒度,导出CSV格式

把oms数据定期取下来，然后上传到odps中创建的结果输出表（OdpsFeeDemo）
tunnel upload C:UsersDesktopaa.txt project.tablename ;
（二）进行表关联，将最终结果存储在上面创建的MaxComputeFee中

select 
distinct 
t.task_schema,
o.MeteringId,
t.owner_id,
o.type,
o.endtime,
o.computationsqlinput,
o.computationsqlcomplexity,
t.cost_cpu,o.starttime,
t.cost_mem 
from information_schema.tasks_history t 
right join OdpsFeeDemo o 
on t.inst_id = o. meteringid 
and t.task_schema = o.projectid
where o.type = “ComputationSql”;

方法2：程序化API下载费用明细数据&上传到MaxCompute后分析
（一）在odps创建oms表OdpsFeeDemo，参考如下：

CREATE TABLE IF NOT EXISTS OdpsFeeDemo(
ProjectId          STRING COMMENT '项目编号',
MeteringId         STRING COMMENT '计量信息编号',
Type                  STRING COMMENT '数据分类',
Storage            STRING COMMENT '存储(Byte)',
EndTime            STRING COMMENT '结束时间',
ComputationSqlInput  STRING COMMENT 'SQL读取量',
ComputationSqlComplexity  STRING COMMENT 'SQL复杂度',
StartTime          STRING COMMENT '开始时间',
OdpsSpecCode       STRING COMMENT '规格类型'
);

通过API下载OMS系统数据并上传到odps对于表格中
代码参考如下：
1) 服务启动类Application

package com.alibaba.odps;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

/**
 * @ClassName: Application
 * @Description: 服务启动类
 * @Author: ***
 * @Data: 2019/7/30 17:15
 **/
@SpringBootApplication
@EnableScheduling
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

2) 从odps接收数据ReceiveData

package com.alibaba.odps.controller;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;
import com.alibaba.fastjson.JSONObject;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.bssopenapi.model.v20171214.QueryUserOmsDataRequest;
import com.aliyuncs.bssopenapi.model.v20171214.QueryUserOmsDataResponse;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.exceptions.ServerException;
import com.aliyuncs.profile.DefaultProfile;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
 * @ClassName: ReceiveData
 * @Description: 接收数据
 * @Author: LiuJianwei
 * @Data: 2019/7/30 17:18
 **/
@Component
public class ReceiveData {

    @Value("${table}")
    private String table;

    @Value("${odps.accessKeyId}")
    private String accessKeyId;

    @Value("${odps.accessKeySecret}")
    private String accessKeySecret;

    @Value("${file.save.path}")
    private String fileSavePath;

    @Autowired
    private OdpsServer odpsServer;

    protected final ObjectMapper objectMapper = new ObjectMapper();

    {
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

//    String[] fileds = {"DBVersion", "InstanceId", "NetworkIn", "NetworkOut", "Storage", "Memory", "Region", "ProviderId",
//           "DBType", "EndTime", "StartTime", "InstanceUseType", "InstanceName"};
    String[] fileds = {"ProjectId","MeteringId","Type","Storage","EndTime","ComputationSqlInput","ComputationSqlComplexity","StartTime","OdpsSpecCode"};@Scheduled(cron = "${cron}")
    public void queryUserOmsData() {
        //获取昨天的开始日期和结束日期
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        String yesterday = format.format(DateUtils.addDays(new Date(), -1));
        //String yesterday = "2019-07-29";
        String startTime = yesterday + "T00:00:00Z";
        String endTime = yesterday + "T23:59:59Z";
        DefaultProfile profile = DefaultProfile.getProfile("cn-hangzhou", accessKeyId, accessKeySecret);
        IAcsClient client = new DefaultAcsClient(profile);
        for (String tab : table.split(",")) {
            QueryUserOmsDataRequest request = new QueryUserOmsDataRequest();
            request.setTable(tab.trim());
            request.setDataType("HOUR");
            request.setStartTime(startTime);
            request.setEndTime(endTime);
            try {
                QueryUserOmsDataResponse response = client.getAcsResponse(request);
                String data = new Gson().toJson(response);
                //将数据插入
                odpsServer.writeDataToOdps(data, yesterday, tab.trim());
                //将查询到的数据保存到TXT中
                writeDataToTxt(data, yesterday);
            } catch (IOException | ServerException e) {
                e.printStackTrace();
            } catch (ClientException e) {
                System.out.println(e);
                System.out.println("ErrCode:" + e.getErrCode());
                System.out.println("ErrMsg:" + e.getErrMsg());
                System.out.println("RequestId:" + e.getRequestId());
            }
        }
    }

    public void writeDataToTxt(String data, String yesterday) throws IOException {
        String path = fileSavePath + File.separator + yesterday + ".txt";
        FileWriter writer = new FileWriter(new File(path));
        if (StringUtils.isNotEmpty(data)) {
            JSONObject json = objectMapper.readValue(data, JSONObject.class);
            JSONObject datas = json.getJSONObject("data");
            if (datas.containsKey("omsData")) {
                List<Map<String, Object>> list = (List<Map<String, Object>>) datas.get("omsData");
                if (!list.isEmpty()) {
                    for (Map<String, Object> map : list) {
                        StringBuilder sb = new StringBuilder();
                        for (String key : fileds) {
                            if (map.containsKey(key)) {
                                sb.append(map.get(key));
                            } else {
                                sb.append(" ");
                            }
                            sb.append(",");
                        }
                        sb.setLength(sb.length() - 1);
                        sb.append("rn");
                        writer.write(sb.toString());
                    }
                }
            }
        }
        writer.flush();
        writer.close();
    }
}

3) 将接收数据上传到MaxCompute项目里建好的oms表，类名：OdpsServer

package com.alibaba.odps.controller;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.odps.Odps;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.RecordWriter;
import com.aliyun.odps.tunnel.TableTunnel;
import com.aliyun.odps.tunnel.TableTunnel.UploadSession;
import com.aliyun.odps.tunnel.TableTunnel.UploadStatus;
import com.aliyun.odps.tunnel.TunnelException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

/**
 * @ClassName: OdpsServer
 * @Description: 将数据写入ODPS中
 * @Author: LiuJianwei
 * @Data: 2019/7/30 17:23
 **/
@Component
public class OdpsServer implements InitializingBean {

    @Value("${odps.accessKeyId}")
    private String accessKeyId;

    @Value("${odps.accessKeySecret}")
    private String accessKeySecret;

    @Value("${odps.project}")
    private String project;

    @Value("${odps.url}")
    private String url;

    private UploadSession ossUploadSession;

    private UploadSession rdsUploadSession;

    private UploadSession odpsUploadSession;

    private String OSSTableName = "MaxComputeFee";

    private String RDSTableName ="RDS";

    private String ODPSTableName ="OdpsFeeDemo";

    protected final ObjectMapper objectMapper = new ObjectMapper();

    {
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    public void writeDataToOdps(String data, String yesday, String tab) {
        List<Map<String, Object>> dataList = new ArrayList<>();
        if (StringUtils.isNotEmpty(data)) {
            try {
                JSONObject json = objectMapper.readValue(data, JSONObject.class);
                JSONObject datas = json.getJSONObject("data");
                if (datas.containsKey("omsData")) {
                    dataList = (List<Map<String, Object>>)datas.get("omsData");
                }
                if (dataList.isEmpty()) {
                    return;
                }
                //数据不为空,开发往ODPS中写入数据
                if (tab.equals("oss")) {
                    for (Map<String, Object> map : dataList) {
                        UploadSession session = getSession(OSSTableName);
                        RecordWriter writer = session.openRecordWriter(session.getAvailBlockId());
                        Record record = session.newRecord();
                        writer.write(record);
                        if (writer != null) {
                            writer.close();
                            session.commit(new Long[] {0L});
                        }
                    }
                } else if (tab.equals("rds")) {
                    for (Map<String, Object> map : dataList) {
                        UploadSession session = getSession(RDSTableName);
                        RecordWriter writer = session.openRecordWriter(session.getAvailBlockId());
                        Record record = session.newRecord();
                        record.set("dbversion", map.get("DBVersion").toString());
                        record.set("instanceid", map.get("InstanceId").toString());
                        record.set("networkin", map.get("NetworkIn").toString());
                        record.set("networkout", map.get("NetworkOut").toString());
                        record.set("storage", Long.valueOf(map.get("Storage").toString()));
                        record.set("memory", map.get("Memory").toString());
                        record.set("region", map.get("Region").toString());
                        record.set("providerid", map.get("ProviderId").toString());
                        record.set("dbtype", map.get("DBType").toString());
                        record.set("endtime", map.get("EndTime").toString());
                        record.set("starttime", map.get("StartTime").toString());
                        record.set("instanceusetype", map.get("InstanceUseType").toString());
                        record.set("instancename", map.get("InstanceName").toString());
                        writer.write(record);
                        if (writer != null) {
                            writer.close();
                            session.commit(new Long[] {0L});
                        }
                    }
                } else if (tab.equals("odps")) {
                    for (Map<String, Object> map : dataList) {
                        UploadSession session = getSession(ODPSTableName);
                        RecordWriter writer = session.openRecordWriter(session.getAvailBlockId());
                        Record record = session.newRecord();
                        record.set("projectid", map.containsKey("ProjectId") ? map.get("ProjectId").toString() : "");
record.set("meteringid", map.containsKey("MeteringId") ? map.get("MeteringId").toString() : "");
record.set("type", map.containsKey("Type") ? map.get("Type").toString() : "");
record.set("storage", map.containsKey("Storage") ? map.get("Storage").toString() : "");
record.set("endtime", map.containsKey("EndTime") ? map.get("EndTime").toString() : "");
record.set("computationsqlinput", map.containsKey("ComputationSqlInput") ? map.get("ComputationSqlInput").toString() : "");
record.set("computationsqlcomplexity", map.containsKey("ComputationSqlComplexity") ? map.get("ComputationSqlComplexity").toString() : "");
record.set("starttime", map.containsKey("StartTime") ? map.get("StartTime").toString() : "");
record.set("odpsspeccode", map.containsKey("OdpsSpecCode") ? map.get("OdpsSpecCode").toString() : "");writer.write(record);
                        if (writer != null) {
                            writer.close();
                            session.commit(new Long[] {0L});
                        }
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException();
            }
        }
    }

    private UploadSession getSession(String tableName) {
        try {
            if (tableName.equals(OSSTableName)) {
                if (!this.ossUploadSession.getStatus().equals(UploadStatus.NORMAL)) {
                    this.ossUploadSession = createNewSession(tableName);
                }
                return this.ossUploadSession;
            } else if (tableName.equals(RDSTableName)) {
                if (!this.rdsUploadSession.getStatus().equals(UploadStatus.NORMAL)) {
                    this.rdsUploadSession = createNewSession(tableName);
                }
                return this.rdsUploadSession;
            }else if (tableName.equals(ODPSTableName)) {
                if (!this.odpsUploadSession.getStatus().equals(UploadStatus.NORMAL)) {
                    this.odpsUploadSession = createNewSession(tableName);
                }
                return this.odpsUploadSession;
            }

        } catch (TunnelException | IOException e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    private UploadSession createNewSession(String tableName) {
        try {
            AliyunAccount account = new AliyunAccount(accessKeyId, accessKeySecret);
            Odps odps = new Odps(account);
            odps.setEndpoint(url);
            odps.setDefaultProject(project);
            TableTunnel odpsTunnel = new TableTunnel(odps);
            UploadSession session = odpsTunnel.createUploadSession(project, tableName);
            return session;
        } catch (TunnelException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.ossUploadSession = createNewSession(OSSTableName);
        this.rdsUploadSession = createNewSession(RDSTableName);
        this.odpsUploadSession = createNewSession(ODPSTableName);
    }

}

4) 配置文件

#配置accessKeyId
odps.accessKeyId=******** 
#配置accessKeySecret
odps.accessKeySecret=********
#配置project
odps.project=工作空间
#配置url
odps.url=http://service.odps.aliyun.com/api
#配置table
table=odps
ds#配置定时任务时间设置
cron=0/1 0/1 * * * ?

5) 现在将数据上传到odps里面对应的表，然后进行关联

select 
distinct 
t.task_schema,
o.MeteringId,
t.owner_id,
o.type,
o.endtime,
o.computationsqlinput,
o.computationsqlcomplexity,
t.cost_cpu,o.starttime,
t.cost_mem 
from information_schema.tasks_history t 
right join OdpsFeeDemo o 
on t.inst_id = o. meteringid 
and t.task_schema = o.projectid
where o.type = “ComputationSql”;

原文链接

本文为云栖社区原创内容，未经允许不得转载。

lisp 读取dwg 统计信息_用阿里云交易和账单管理API实现MaxCompute费用对账分摊统计...相关推荐

  1. 优化器统计信息_高水位_柱状图等

    本文章为网络笔记,看了warehouse老师的视频受益匪浅,更是感觉自己技术太过初级,特写了本笔记,方便以后反复学习! 如有任何不妥,请发邮件至102448567@qq.com删除文章! 关于ware ...

  2. 基于Javaweb和阿里云服务器的用户管理平台

    基于Javaweb和阿里云服务器的用户管理平台 技术支持: 视图层:js+html+css 业务层:servlet 持久层:mysql5.6 项目介绍:myEclipse负责开发javaweb项目,t ...

  3. 阿里云code结合git管理代码，运用webHook同步部署服务器代码（php）

    项目场景: 阿里云code结合git管理代码,运用webHook同步部署服务器代码(php) 问题描述: git生成密钥时linux用户为root,php运行用户是www.结果导致在服务器端git p ...

  4. JAVA实现对阿里云DNS的解析管理

    1.阿里云DNS的SDK依赖 <dependency> <groupId>com.aliyun</groupId> <artifactId>tea-op ...

  5. python读取微信群信息_利用Python转发微信群通知信息

    案例:作为班长需要实时转发年级群发布的有关通知.文件到自己班的微信群,这种无脑的重复性工作适用于编写脚本自动帮忙处理. 准备:服务器(可选择阿里云学生服务器).itchat模块.一颗爱折腾的心 代码: ...

  6. oss多线程 上传_解读阿里云oss-android/ios-sdk 断点续传（多线程）

    摘要: oss sdk 断点续传功能使用及其相关原理 前言 移动端现状 随着移动端设备的硬件水平的不断提高,如今的cpu,内存等方面都大大的超过了一般的pc电脑,因此在现今的程序中,合理的使用多线程去 ...

  7. 【阿里云MVP月度分享】SaaS服务商如何通过数加平台统计业务流量

    摘要: 一.概述 因为自家公司是做B2B类Saas服务的,难免会产生精准计费的问题,所以在通过多套方案的选型及对比以后,我们最终确定了以下的方式进行自有业务平台的流量计算方案.因为涉及到具体的操作,所 ...

  8. CTU-UHB Intrapartum Cardiotocography Database 产时CTG信号数据下载与读取（2021.7.9 新增阿里云盘下载）

    目录 CTU-CHB Abstract Data Description Acknowledgements 正文 数据读取 MATLAB代码 数据整理 注意! CTU-CHB 首先说明一下什么是CTU ...

  9. 阿里云mysql数据库引擎_在阿里云RDS数据库服务器中将MySQL InnoDB引擎表转为压缩格式...

    今年以来将我们以前托管或者租用的服务器全面转向阿里云,除了采用ECS服务器以外,还有一项重要的是采用了RDS数据库服务器,这对于服务的稳定性.各项指标的监控.调优等都有帮助. 不过随着近期更多数据库转 ...

最新文章

  1. 简单快捷方法：CAJ 文件转 成PDF文件
  2. 贵港天气预报软件测试,贵港天气预报15天
  3. 【已解决】查看Python中已经安装的包
  4. 线性支持向量机完全理解版
  5. 在php中函数名称区分大小写吗,PHP中的变量名，函数名，类名是区分大小写的吗...
  6. mac使用Shell(终端)SSH连接远程服务器
  7. python开源项目贡献_如何通过为开源项目做贡献来发展自己的职业
  8. 介绍struct bus_type非常好的
  9. Arduino 系列传感器应用
  10. 计算机学报范文,计算机学报论文
  11. PHP小白之路1--PHP之简易留言板设计
  12. java 数组 重复数据结构_Java数据结构和算法( 二 ) ## 数组
  13. 活着就是要做有意义的事,做有意义的事就是好好活着
  14. P1779 小胡同学的跳板
  15. Nood.js C++ 模块 addon插件（二）
  16. 人月神话（五）未雨绸缪、干将莫邪、整体部分
  17. windows日志捕获工具-DebugView使用教程
  18. 你是在独立思考，还是在被洗脑？
  19. 直方图匹配算法设计文档
  20. 射频混频器matlab,一个很多射频/天线工程师不知道的神奇网站

热门文章

  1. [转载] Dreamweaver——02 创建网页的基本对象
  2. 布同：pdf自定义分割（断章）
  3. 我和Django那些事儿（8）----相册django插件photologue，jQuery插件Slides
  4. Mongodb常规操作【一】
  5. java中实现全局变量的功能
  6. CS100.1x Introduction to Big Data with Apache Spark
  7. Android emulator error: x86 emulation currently requires hardware acceleration的解决方案
  8. 汇编--查找第一个非0字符的五种方法
  9. 虚拟机可以识别usb无线网卡，但一直提示设备正在运行中--解决方法
  10. 德勤元宇宙综观：愿景、技术和应对