MaxCompute计算作业成本控制_云原生大数据计算服务 MaxCompute(MaxCompute)-阿里云帮助中心

当您发现MaxCompute账单持续上涨，而且成本变得难以管理时，您可以从计算作业着手，通过对SQL作业和MapReduce作业的优化而减少计算成本。本文为您介绍SQL作业和MapReduce作业计算成本的控制方法。

预估计算成本

您可以在计算前对计算成本进行预估，控制计算成本。详细的预估方法，请参见TCO工具。也可以配置消费预警，预防意料之外的高额费用。预警方法设置请参见消费监控告警。如果计算成本过高，您可以参考下面的方法进行优化，以控制计算成本。

SQL作业计算成本控制

对于SQL计算作业，大部分费用较高的SQL都是由全表扫描引起的。另外，调度频繁也会引起SQL作业费用的增加，调度频繁可能会产生任务的堆积，在后付费的情况下会造成排队现象，如果任务多又出现了排队，那么第二天的账单就会异常。通过如下策略进行SQL作业计算成本控制：

避免频繁调度。MaxCompute是批量计算的服务，距离实时的计算服务还是存在一定距离的。如果间隔时间变短，计算频率增加，再加上使用SQL的不良习惯就会导致计算费用飙升，产生费用较高的账单。所以请尽量避免频繁调度，如果要进行频繁调度请通过CostSQL等方式预估一下SQL的开销到底有多大，不然会造成较大预估外的开销。
控制全表扫描。您可以通过以下几种策略来控制全表扫描问题：
- 设置参数关闭全表扫描功能。目前支持Session级别和Project级别的控制。
```
--禁止session 级别全表扫描。
set odps.sql.allow.fullscan=false;
--禁止project级别全表扫描。
SetProject odps.sql.allow.fullscan=false;
```
- 使用列剪裁。在读数据的时候，只读取查询中需要用到的列，而忽略其他列，避免使用SELECT *引起全表扫描。
```
SELECT a,b FROM T WHERE e < 10;
```
  其中，T包含5个列(a,b,c,d,e)，列c，d将会被忽略，只会读取a，b，e列。
- 使用分区剪裁。分区剪裁是指对分区列指定过滤条件，使得只读取表的部分分区数据，避免全表扫描引起的错误及资源浪费。
```
SELECT a,b FROM T WHERE partitiondate='2017-10-01';
```
- SQL关键字的优化。计费的SQL关键字包括：JOIN、GROUP BY、ORDER BY、DISTINCT、INSERT INTO。您可以根据以下建议进行优化：
  - 在进行JOIN的时候，一定要先进行分区剪裁再进行JOIN，不然的话就可能会先做全表扫描。分区裁剪失效请参考分区剪裁失效的场景分析。
  - 减少FULL OUTER JOIN 的使用，改为UNION ALL。
```
SELECT COALESCE(t1.id, t2.id) AS id, SUM(t1.col1) AS col1
 , SUM(t2.col2) AS col2
FROM (
 SELECT id, col1
 FROM table1
) t1
FULL OUTER JOIN (
 SELECT id, col2
 FROM table2
) t2
ON t1.id = t2.id
GROUP BY COALESCE(t1.id, t2.id);
--可以优化为如下语句。
SELECT t.id, SUM(t.col1) AS col1, SUM(t.col2) AS col2
FROM (
 SELECT id, col1, 0 AS col2
 FROM table1
 UNION ALL
 SELECT id, 0 AS col1, col2
 FROM table2
) t
GROUP BY t.id;
```
  - 在UNION ALL内部尽可能不使用GROUP BY，改为在外层统一GROUP BY。
```
SELECT t.id, SUM(t.val) AS val
FROM (
 SELECT id, SUM(col3) AS val
 FROM table3
 GROUP BY id
 UNION ALL
 SELECT id, SUM(col4) AS val
 FROM table4
 GROUP BY id
) t
GROUP BY t.id;
可以优化为---------------------------
SELECT t.id, SUM(t.val) AS val
FROM (
 SELECT id, col3 AS val
 FROM table3
 UNION ALL
 SELECT id, col4 AS val
 FROM table4
) t
GROUP BY t.id;
```
  - 临时导出的数据如果需要排序，尽量在导出后使用Excel等工具进行排序，避免使用ORDER BY。
  - 尽量避免使用DISTINCT关键字，改为多套一层GROUP BY。
```
SELECT COUNT(DISTINCT id) AS cnt
FROM table1;
可以优化为---------------------------
SELECT COUNT(1) AS cnt
FROM (
 SELECT id
 FROM table1
 GROUP BY id
) t;
```
  - 尽量避免使用INSERT INTO方式写入数据，可以考虑增加一个分区字段。通过降低SQL复杂度，来节省SQL的费用。
避免使用运行查询的方式预览表数据。如果您想预览表数据，可以使用表预览的方式查看数据，而不会产生费用。如果您使用DataWorks，在数据地图页面，可以预览表以及查看表的详情，具体方法请参见查看表详情。如果您使用MaxCompute Studio，双击表就可以进行表数据预览。
计算时合理的选择工具。由于MaxCompute的查询响应是分钟级，不适合直接用于前端查询，计算出的结果数据同步到外部存储中保存，对于大部分用户来说，关系型数据库是最优先的选择。轻度计算推荐使用MaxCompute，重度计算（即直接出最终结果。前端展示时，不做任何判断、聚合、关联字典表、甚至不带WHERE条件）推荐使用RDS等关系型数据库。

MapReduce作业计算成本控制

通过如下策略进行MapReduce作业计算成本控制：

设置合理的参数。
- split size
  Map默认的split size是256 MB，split size的大小决定了Map个数多少，如果用户的代码逻辑比较耗时，Map需要较长时间结束，可以通过JobConf#setSplitSize方法适当调小split size。然而split size也不宜设置太小，否则会占用过多的计算资源。
- MapReduce Reduce Instance
  单个job默认Reduce Instance个数为Map Instance个数的1/4，用户设置作为最终的Reduce Instance个数，范围[0, 2000]，数量越多，计算时消耗越多，成本越高，应合理设置。
MapReduce减少中间环节
如果有多个MapReduce作业之间有关联关系，前一个作业的输出是后一个作业的输入，可以考虑采用Pipeline的模式，将多个串行的MapReduce作业合并为一个，这样可以用更少的作业数量完成同样的任务。一方面减少中间表造成的多余磁盘IO，提升性能；另一方面减少作业数量使调度更加简单，增强流程的可维护性，具体使用方法请参见Pipeline示例。
对输入表列裁剪
对于列数特别多的输入表，Map阶段处理只需要其中的某几列，可以通过在添加输入表时明确指定输入的列，减少输入量。例如只需要c1，c2列，可以参考如下设置。
```
InputUtils.addTable(TableInfo.builder().tableName("wc_in").cols(new String[]{"c1","c2"}).build(), job);
```
设置后，在Map中读取到的Record就只有c1，c2列，如果之前是使用列名获取Record数据，不会有影响，而用下标获取的需要注意这个变化。
避免资源重复读取
资源的读取尽量放置到Setup阶段读取，避免资源多次读取的性能损失，另外系统也有64次读取的限制，资源的读取请参见使用资源示例。

减少对象构造开销

对于Map、Reduce阶段每次都会用到的Java对象，避免在Map/Reduce函数里构造，可以放到Setup阶段，避免多次构造产生的开销。

{
    ...
    Record word;
    Record one;

    public void setup(TaskContext context) throws IOException {


      // 创建一次就可以，避免在map中每次重复创建。
      word = context.createMapOutputKeyRecord();

      one = context.createMapOutputValueRecord();

      one.set(new Object[]{1L});

    }
    ...
}

合理使用Combiner

如果Map的输出结果中有很多重复的Key，可以合并后输出，Combiner后可以减少网络带宽传输和一定Shuffle的开销。如果Map输出本来就没有多少重复的，就不要用Combiner，用了反而可能会有一些额外的开销。Combiner实现的是和Reducer相同的接口，例如一个WordCount程序的Combiner可以定义如下。

/**
   * A combiner class that combines map output by sum them.
   */
  public static class SumCombiner extends ReducerBase {

    private Record count;

    @Override
    public void setup(TaskContext context) throws IOException {
      count = context.createMapOutputValueRecord();
    }

    @Override
    public void reduce(Record key, Iterator<Record> values, TaskContext context)
        throws IOException {
      long c = 0;
      while (values.hasNext()) {
        Record val = values.next();
        c += (Long) val.get(0);
      }
      count.set(0, c);
      context.write(key, count);
    }
  }

合理选择Partition Column或自定义Partitioner
合理选择Partition Columns，可以使用JobConf#setPartitionColumns这个方法进行设置（默认是Key Schema定义的Column），设置后数据将按照指定的列计算HASH值分发到Reduce中，避免数据倾斜导致作业长尾现象，如有必要也可以选择自定义Partitioner，自定义Partitioner的使用方法如下。
```
import com.aliyun.odps.mapred.Partitioner;

public static class MyPartitioner extends Partitioner {

@Override
public int getPartition(Record key, Record value, int numPartitions) {
  // numPartitions即对应reducer的个数
  // 通过该函数决定map输出的key value去往哪个reducer。
  String k = key.get(0).toString();
  return k.length() % numPartitions;
}
}
```
在jobconf里进行设置如下。
```
jobconf.setPartitionerClass(MyPartitioner.class)
```
需要在jobconf里明确指定Reducer的个数。
```
jobconf.setNumReduceTasks(num)
```
合理使用JVM内存参数
过于追求调优，把MapReduce任务内存设置过大也会造成成本上升。标准配置是1 Core 4G ，odps.stage.reducer.jvm.mem=4006，当CPU与内存比超过1:4时，对应的费用也会大幅升高。

预估计算成本

SQL作业计算成本控制

MapReduce作业计算成本控制

相关文档