MaxComputeコンピューティングジョブのコスト管理 - MaxCompute

MaxComputeの請求額が絶えず増加し、コストの管理が困難になった場合は、SQLジョブとMapReduceジョブを最適化してコンピューティングコストを削減できます。このトピックでは、SQLジョブとMapReduceジョブのコンピューティングコストを制御する方法について説明します。

計算コストの見積もり

コンピューティングジョブを実行する前に、コンピューティングコストを見積もることができます。詳細については、「課金方法の選択」トピックのTCOツールをご参照ください。リソース消費のアラートを設定して、余分なコストが発生しないようにすることもできます。

コンピューティングコストが高い場合は、このトピックで説明されている方法を使用してコストを削減できます。

SQLジョブのコンピューティングコストを制御する

フルテーブルスキャンをトリガーする一部のSQLジョブは、高いコンピューティングコストが発生します。 SQLジョブの頻繁なスケジューリングは、ジョブの蓄積を引き起こす可能性があり、これもコンピューティングコストを増加させる。累積が発生し、従量課金が使用されている場合、ジョブはキューに入れられ、より多くのリソースが必要になります。その結果、翌日に発生する請求書は異常に高くなります。次の方法を使用して、SQLジョブのコンピューティングコストを制御できます。

頻繁なスケジューリングを避ける。 MaxComputeは、大量のデータを一度に処理するコンピューティングサービスを提供します。これは、リアルタイムコンピューティングサービスとは異なります。 SQLジョブが短い間隔で実行されると、計算頻度が増加します。コンピューティング頻度の増加とSQLジョブの不適切な実行は、コンピューティングコストの増加を引き起こします。頻繁なスケジューリングが必要な場合は、CostSQLを使用してSQLジョブのコストを見積もり、追加コストを回避します。
テーブル全体のスキャンを減らす。次の方法を使用できます。
- テーブル全体のスキャン機能を無効にするために必要なパラメーターを指定します。セッションまたはプロジェクトの機能を無効にできます。
```
-- Disable the feature for a session. 
set odps.sql.allow.fullscan=false;
-- Disable the feature for a project. 
SetProject odps.sql.allow.fullscan=false;
```
- 列を剪定します。列のプルーニングにより、システムは必要な列からのみデータを読み取ることができます。 SELECT * ステートメントは使用しないことを推奨します。これは、ステートメントを実行するとテーブル全体のスキャンがトリガーされるためです。
```
SELECT a,b FROM T WHERE e < 10;
```
  このステートメントでは、Tテーブルにa、b、c、d、およびeの列が含まれます。ただし、a、b、e列のみが読み取られます。
- パーティションを剪定します。パーティションプルーニングでは、パーティションキー列のフィルタ条件を指定できます。このようにして、システムは必要なパーティションからのみデータを読み取ります。これにより、テーブル全体のスキャンによるエラーやリソースの浪費が回避されます。
```
SELECT a,b FROM T WHERE partitiondate='2017-10-01';
```
- コストが発生するSQLキーワードを最適化します。キーワードには、JOIN、GROUP BY、ORDER BY、DISTINCT、INSERT INTOが含まれます。次のルールに基づいてキーワードを最適化できます。
  - 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);
-- Optimized: 
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を含めないでください。 UNION ALLの外部で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;
Optimized:---------------------------
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;
```
  - 一時的にエクスポートしたデータを並べ替えるには、ORDER byの代わりにEXCELなどのツールを使用してデータを並べ替えます。
  - DISTINCTを使用しないでください。代わりにGROUP BYを使用します。
```
SELECT COUNT(DISTINCT id) AS cnt
FROM table1;
Optimized:---------------------------
SELECT COUNT(1) AS cnt
FROM (
 SELECT id
 FROM table1
 GROUP BY id
) t;
```
  - データの書き込みにINSERT INTOを使用しないでください。代わりにパーティションフィールドを追加します。これにより、SQLの複雑さが軽減され、コンピューティングコストが節約されます。
SQL文を実行してテーブルデータをプレビューしないでください。テーブルプレビュー機能を使用して、テーブルデータを表示できます。この方法は、コンピューティングリソースを消費しません。 DataWorksを使用している場合は、[データマップ] ページでテーブルをプレビューし、テーブルの詳細を照会できます。詳細については、「テーブルの詳細の表示」をご参照ください。 MaxCompute Studioを使用する場合は、テーブルをダブルクリックしてデータをプレビューします。
データコンピューティングに適したツールを選択します。 MaxComputeは数分以内にクエリに応答します。フロントエンドクエリには適していません。計算結果は、外部ストレージシステムに同期される。ほとんどのユーザーは、リレーショナルデータベースを使用して結果を保存します。 MaxComputeを軽量コンピューティングジョブに使用し、ApsaraDB for RDSなどのリレーショナルデータベースをフロントエンドクエリに使用することを推奨します。フロントエンドクエリには、クエリ結果のリアルタイム生成が必要です。クエリ結果がフロントエンドに表示されている場合、データに対して条件句は実行されません。データは、辞書に集約または関連付けられていません。クエリにはWHERE句も含まれていません。

MapReduceジョブのコンピューティングコストを制御する

次の方法を使用して、MapReduceジョブのコンピューティングコストを制御できます。

必要な設定を構成するConfigure the required settings
- 分割サイズ
  マッパーのデフォルトの分割サイズは256 MBです。分割サイズは、マッパーの数を決定します。マッパーのコードロジックに時間がかかる場合は、JobConf#setSplitSizeを使用してsplit sizeを縮小できます。適切なsplit sizeを設定する必要があります。そうでなければ、過剰なコンピューティングリソースが消費され得る。
- MapReduceインスタンス
  デフォルトでは、ジョブを完了するために使用されるリデューサーの数は、マッパーの数の4分の1です。レデューサーの数を0から2,000の範囲の値に設定できます。より多くの減速機はより多くのコンピューティングリソースを必要とし、コストが増加します。レデューサーの数を適切に設定する必要があります。
MapReduceジョブの数を減らす
複数のMapReduceジョブが関連付けられており、ジョブの出力が次のジョブの入力である場合は、パイプラインモードを使用することを推奨します。パイプラインモードでは、複数のシリアルMapReduceジョブを1つのジョブにマージできます。これにより、中間テーブルによって引き起こされる冗長なディスクI/O操作が削減され、パフォーマンスが向上します。これはまた、ジョブスケジューリングを簡略化し、プロセス保守効率を高める。詳細については、「パイプラインの例」をご参照ください。
入力テーブルの列を剪定する
多数の列を含む入力テーブルの場合、少数の列のみがマッパーによって処理される。入力テーブルを追加するときに、列を指定して、読み取る必要のあるデータ量を減らすことができます。たとえば、c1列とc2列のデータを処理するには、次の設定を使用します。
```
InputUtils.addTable(TableInfo.builder().tableName("wc_in").cols(new String[]{"c1","c2"}).build(), job);
```
設定後、マッパはc1列とc2列からのみデータを読み取ります。これは、列名に基づいて取得されるデータには影響しません。ただし、これは下付き文字に基づいて取得されるデータに影響を与える可能性があります。
リソースの重複読み取りを避ける
セットアップ段階でリソースを読むことを推奨します。これにより、重複するリソース読み取りによるパフォーマンスの低下を防ぎます。リソースを最大64回読み取ることができます。詳細については、「リソースの使用例」をご参照ください。
オブジェクト構築のオーバーヘッドを減らす
mapおよびreduceステージで毎回使用されるJavaオブジェクトの場合は、mapまたはreduce関数でそれらを構築しないようにします。複数の構造によって引き起こされるオーバーヘッドを防ぐために、それらをセットアップ段階に置くことができます。
```
{
    ...
    Record word;
    Record one;

    public void setup(TaskContext context) throws IOException {


      // Create a Java object in the setup stage. This prevents the repeated creation of Java objects in each map stage. 
      word = context.createMapOutputKeyRecord();

      one = context.createMapOutputValueRecord();

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

    }
    ...
}
```
適切な方法でコンバイナを使用する
マップタスクの出力に複数の重複キーが含まれている場合は、コンバイナを使用してこれらのキーをマージできます。これにより、伝送帯域幅とシャッフルのオーバーヘッドが減少します。マップタスクの出力が複数の重複キーを含まない場合、コンバイナを使用すると余分なオーバーヘッドが発生する可能性があります。コンバイナは、リデューサインタフェースを実装する。次のコードは、WordCountプログラムのコンバイナを定義します。
```
/**
   * 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);
    }
  }
```
パーティションキー列を適切に選択するか、パーティショナーをカスタマイズする
JobConf#setPartitionColumnsを使用して、パーティションキー列を指定できます。デフォルトのパーティションキー列は、キースキーマで定義されます。この方法を使用すると、指定された列のハッシュ値に基づいてデータがリデューサに転送されます。これにより、データスキューによるロングテールの問題が防止されます。必要に応じて、パーティショナーをカスタマイズすることもできます。次のコードは、パーティショナーをカスタマイズする方法を示しています。
```
import com.aliyun.odps.mapred.Partitioner;

public static class MyPartitioner extends Partitioner {

@Override
public int getPartition(Record key, Record value, int numPartitions) {
  // numPartitions indicates the number of reducers.
  // This function is used to determine the reducers to which the keys of map tasks are transferred. 
  String k = key.get(0).toString();
  return k.length() % numPartitions;
}
}
```
jobconfで次の設定を行います。
```
jobconf.setPartitionerClass(MyPartitioner.class)
```
jobconfのリデューサーの数を指定します。
```
jobconf.setNumReduceTasks(num)
```
必要に応じてJVMメモリパラメータを設定する
MapReduceジョブのメモリが大きいため、コンピューティングコストが増加します。標準設定では、1 CPUコアと4 GBのメモリが設定され、odps.stage.reducer.jvm.memは4006に設定されています。 CPUコアとメモリの比率が大きい (1:4を超える) と、コンピューティングコストも増加します。

MaxCompute:コンピューティングコストの制御

計算コストの見積もり

SQLジョブのコンピューティングコストを制御する

MapReduceジョブのコンピューティングコストを制御する

関連ドキュメント