MaxCompute:MaxCompute での上位 N 件の請求アカウントと時間のかかるジョブの分析

ステップ 1：Information Schema サービスの取得

2024 年 3 月 1 日以降、MaxCompute はデフォルトで新規プロジェクトに対してプロジェクトレベルの Information Schema パッケージを自動的にインストールしなくなりました。メタデータをクエリする必要があるサービスがある場合は、テナントレベルの Information Schema をクエリして、より完全な情報を取得できます。テナントレベルの Information Schema の使用方法の詳細については、「テナントレベルの Information Schema」をご参照ください。

(オプション) ステップ 2：プロジェクトオーナー以外のユーザーへの権限付与

Information Schema ビューには、プロジェクトレベルのすべてのユーザーデータが含まれています。デフォルトでは、プロジェクトオーナーがこのデータを表示できます。プロジェクト内の他のユーザーやロールが表示する必要がある場合は、権限を付与する必要があります。詳細については、「パッケージに基づいてプロジェクト間でリソースにアクセスする」をご参照ください。

grant <actions> on package Information_Schema.systables to user <user_name>;
grant <actions> on package Information_Schema.systables to role <role_name>;

grant read on package Information_Schema.systables to user RAM$Bob@aliyun.com:user01;

ステップ 3：メタデータのダウンロードとバックアップ

MaxCompute プロジェクトにメタデータバックアップテーブルを作成し、定期タスクをスケジュールしてメタデータを書き込みます。次の手順では、MaxCompute クライアントを使用した例を示します。

1. MaxCompute クライアントにログインし、次のコマンドを実行してメタデータバックアップテーブルを作成します。

   -- project_name は、ご利用の MaxCompute プロジェクトの名前です。
   CREATE TABLE if NOT EXISTS <project_name>.information_history
   (
       task_catalog STRING
       ,task_schema STRING
       ,task_name STRING
       ,task_type STRING
       ,inst_id STRING
       ,`status` STRING
       ,owner_id STRING
       ,owner_name STRING
       ,result STRING
       ,start_time DATETIME
       ,end_time DATETIME
       ,input_records BIGINT
       ,output_records BIGINT
       ,input_bytes BIGINT
       ,output_bytes BIGINT
       ,input_tables STRING
       ,output_tables STRING
       ,operation_text STRING
       ,signature STRING
       ,complexity DOUBLE
       ,cost_cpu DOUBLE
       ,cost_mem DOUBLE
       ,settings STRING
       ,ds STRING
   );

2. DataWorks のデータスタジオインターフェイスに移動し、ODPS SQL ノード (information_history) を作成し、時間指定スケジューリングを構成して定期的に information_history バックアップテーブルにデータを書き込みます。次に、左上隅の アイコンをクリックしてノードを保存します。

   詳細については、「ODPS SQL ノードの作成」をご参照ください。

   次のコードは、ODPS SQL ノードで実行するコマンドの例を示しています。

   -- project_name は、MaxCompute プロジェクトの名前です。
   use <project_name>;
   insert into table <project_name>.information_history select * from information_schema.tasks_history where ds ='datetime1';

   ${datetime1} は DataWorks のスケジューリングパラメーターです。ODPS SQL ノードの右側で、[スケジューリング設定] をクリックします。[基本プロパティ] セクションで、[パラメーター] を datetime1=${yyyymmdd} に設定します。

   説明

   複数の MaxCompute プロジェクトのメタデータを同時に分析するには、複数の ODPS SQL ノードを作成します。各ノードを構成して、異なるプロジェクトのメタデータを同じバックアップテーブルに書き込みます。

ステップ 4：上位 N 件の請求アカウントと時間のかかるジョブを分析するジョブの作成

TASKS_HISTORY ビューの settings フィールドには、スケジューラまたはユーザーからの情報が JSON 形式で記録されます。これには、useragent、bizid、skynet_id、skynet_nodename が含まれます。settings フィールドを使用して、ジョブを作成した RAM ユーザーを見つけることができます。これにより、バックアップテーブルを使用して、上位 N 件の請求アカウントと時間のかかるジョブを見つけることができます。手順は次のとおりです。

1. MaxCompute クライアントにログインし、user_ram という名前の RAM ユーザー詳細テーブルを作成します。このテーブルには、分析対象のアカウントとアカウント ID が記録されます。

   次のコードは、コマンドの例を示しています。

   CREATE TABLE if NOT EXISTS <project_name>.user_ram
   (
       user_id STRING
       ,user_name STRING
   );

2. cost_topn という名前の詳細テーブルを作成して、上位 N 件の請求アカウントの詳細を記録します。

   次のコードは、コマンドの例を示しています。

   CREATE TABLE if NOT EXISTS <project_name>.cost_topn
   (
       cost_sum DECIMAL(38,5)
       ,task_owner STRING
   )
   partitioned BY 
   (
       ds STRING
   );

3. time_topn という名前の詳細テーブルを作成して、上位 N 件の時間のかかるジョブの詳細を記録します。

   次のコードは、コマンドの例を示しています。

   CREATE TABLE if NOT EXISTS <project_name>.time_topn
   (
       inst_id STRING
       ,cost_time BIGINT
       ,task_owner STRING
   )
   partitioned BY 
   (
       ds STRING
   );

4. DataWorks のデータスタジオインターフェイスに移動します。ODPS SQL ノード (topn) を作成し、時間指定スケジューリングを構成して、`cost_topn` テーブルの統計データを `user_ram` テーブルに定期的に書き込みます。次に、左上隅の アイコンをクリックしてノードを保存します。

   詳細については、「ODPS SQL ノードの作成」をご参照ください。

   次のコードは、ODPS SQL ノードで実行するコマンドの例を示しています。

   -- データ型 2.0 を有効にします。データ型 2.0 の詳細については、「データ型のエディション」をご参照ください。
   SET odps.sql.decimal.odps2=true;
   
   -- メタデータを cost_topn および time_topn テーブルに書き込みます。user_id はアカウント ID です。アカウント ID は個人情報ページで確認できます。
   INSERT INTO TABLE <project_name>.cost_topn PARTITION (ds = '${datetime1}')
   SELECT   
       NVL(cost_sum,0) cost_sum
       ,CASE WHEN a.task_owner='<user_id>' OR a.task_owner='<user_id>' OR a.task_owner='<user_id>' THEN b.user_name 
             ELSE a.task_owner 
        END task_owner 
   FROM    (
               SELECT  inst_id
                       ,owner_name
                       ,task_type
                       ,a.input_bytes
                       ,a.cost_cpu
                       ,a.status
                       ,CASE    WHEN a.task_type = 'SQL' THEN CAST(a.input_bytes/1024/1024/1024 * a.complexity * 0.3 AS DECIMAL(18,5) )
                                WHEN a.task_type = 'SQLRT' THEN CAST(a.input_bytes/1024/1024/1024 * a.complexity * 0.3 AS DECIMAL(18,5) )
                                WHEN a.task_type = 'CUPID' AND a.status='Terminated' THEN CAST(a.cost_cpu/100/3600 * 0.66 AS DECIMAL(18,5) ) 
                                ELSE 0 
                        END cost_sum
                       ,a.settings
                       ,GET_JSON_OBJECT(settings, "$.SKYNET_ONDUTY") owner
                       ,CASE    WHEN GET_JSON_OBJECT(a.settings, "$.SKYNET_ONDUTY") IS NULL THEN owner_name 
                                ELSE GET_JSON_OBJECT(a.settings, "$.SKYNET_ONDUTY") 
                        END task_owner
               FROM    information_history
               WHERE   ds = '${datetime1}'
           ) a
   LEFT JOIN <project_name>.user_ram b
   ON      a.task_owner = b.user_id;
   
   INSERT INTO TABLE <project_name>.time_topn PARTITION(ds = '${datetime1}')
   SELECT  inst_id
           ,cost_time
           ,CASE    WHEN a.task_owner='<user_id>' OR a.task_owner='<user_id>' OR a.task_owner='<user_id>' THEN b.user_name 
                    ELSE a.task_owner 
            END task_owner
   FROM    (
               SELECT  inst_id
                       ,task_type
                       ,status
                       ,DATEDIFF(a.end_time, a.start_time, 'ss') AS cost_time
                       ,CASE    WHEN GET_JSON_OBJECT(a.settings, "$.SKYNET_ONDUTY") IS NULL THEN owner_name 
                                ELSE GET_JSON_OBJECT(a.settings, "$.SKYNET_ONDUTY") 
                        END task_owner
               FROM    <project_name>.information_history a
               WHERE   ds = '${datetime1}'
           ) a
   LEFT JOIN <project_name>.user_ram b
   ON      a.task_owner = b.user_id;

   説明

   この例では、task_type = 'SQL' は SQL ジョブ、task_type = 'SQLRT' はクエリアクセラレーションジョブ、task_type = 'CUPID' は Spark ジョブを表します。MapReduce や Mars ジョブなど、他の課金対象ジョブを分析するには、それぞれの課金計算式に基づいてコード行を追加します。課金の詳細については、「コンピューティングユニットベースの課金 (従量課金)」をご参照ください。

   ${datetime1} は DataWorks のスケジューリングパラメーターです。ODPS SQL ノードの右側で、[スケジューリング設定] をクリックします。[基本プロパティ] セクションで、[パラメーター] を datetime1=${yyyymmdd} に設定します。

ステップ 5：DingTalk グループチャットボットの作成と、上位 N 件の請求アカウントおよび時間のかかるジョブに関する情報のプッシュ

以下の手順では、DingTalk グループチャットボットを作成し、上位 N 件の請求アカウントと時間のかかるジョブに関する情報をプッシュする方法を説明します。この手順では、DingTalk PC クライアントを例として使用します。

1. DingTalk グループチャットボットを作成します。

   1. 対象の DingTalk グループを選択し、右上隅の アイコンをクリックします。

   2. [グループ設定] パネルで、[スマートグループアシスタント] をクリックします。

   3. [スマートグループアシスタント] パネルで、[ロボットを追加] をクリックします。

   4. [グループボット] ダイアログボックスの [ロボットを追加] セクションで、 アイコンをクリックします。

   5. [グループボット] ダイアログボックスで、[カスタム] をクリックします。

   6. [ロボットの詳細] ダイアログボックスで、[追加] をクリックします。

   7. [ロボットを追加] ダイアログボックスで、ロボット情報を編集します。

      プロパティ名	ルールの設定
      プロフィール写真	プロフィール写真の右下隅にある アイコンをクリックして編集します。
      ロボット名	ロボットの名前を入力します。
      セキュリティ設定	必要なセキュリティ設定を構成し (少なくとも 1 つ選択)、[「カスタムロボットサービスと免責事項」を読み、同意します] を選択して、[完了] をクリックします。 セキュリティ設定には 3 つのタイプがあります。 カスタムキーワード：最大 10 個のキーワードを設定できます。 署名を追加：[署名を追加] を選択して、ロボットのキーを取得します。 IP アドレス (範囲)：指定された IP アドレス範囲からのリクエストのみが処理されます。

   8. [ロボットを追加] ダイアログボックスで、生成された Webhook URL をコピーします。その後、[完了] をクリックします。

      重要

      Webhook URL は安全に保管してください。外部のウェブサイトに投稿しないでください。URL が漏洩すると、セキュリティリスクが発生する可能性があります。

2. IntelliJ IDEA を使用して Maven プロジェクトを作成し、メッセージを DingTalk グループにプッシュする Java プログラムをコンパイルします。コンパイル後、JAR パッケージが生成されます。

   IntelliJ IDEA の使用方法の詳細については、IntelliJ IDEA インターフェイスの右上隅にある [ヘルプ] をクリックしてください。

   1. Pom 依存関係を構成します。

      次のコードは Pom 依存関係を示しています。

      <?xml version="1.0" encoding="UTF-8"?>
      <project xmlns="http://maven.apache.org/POM/4.0.0"
               xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
               xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
          <modelVersion>4.0.0</modelVersion>
      
          <groupId>DingTalk_Information</groupId>
          <artifactId>DingTalk_Information</artifactId>
          <version>1.0-SNAPSHOT</version>
          <dependencies>
              <dependency>
                  <groupId>com.aliyun.odps</groupId>
                  <artifactId>odps-sdk-core</artifactId>
                  <version>0.35.5-public</version>
              </dependency>
              <dependency>
                  <groupId>log4j</groupId>
                  <artifactId>log4j</artifactId>
                  <version>1.2.15</version>
                  <exclusions>
                      <exclusion>
                          <groupId>com.sun.jmx</groupId>
                          <artifactId>jmxri</artifactId>
                      </exclusion>
                      <exclusion>
                          <groupId>com.sun.jdmk</groupId>
                          <artifactId>jmxtools</artifactId>
                      </exclusion>
                      <exclusion>
                          <groupId>javax.jms</groupId>
                          <artifactId>jms</artifactId>
                      </exclusion>
                  </exclusions>
              </dependency>
              <dependency>
                  <groupId>com.aliyun</groupId>
                  <artifactId>alibaba-dingtalk-service-sdk</artifactId>
                  <version>1.0.1</version>
              </dependency>
              <dependency>
                  <groupId>com.aliyun.odps</groupId>
                  <artifactId>odps-jdbc</artifactId>
                  <version>3.0.1</version>
                  <classifier>jar-with-dependencies</classifier>
              </dependency>
          </dependencies>
          <build>
              <plugins>
                  <plugin>
                      <groupId>org.apache.maven.plugins</groupId>
                      <artifactId>maven-assembly-plugin</artifactId>
                      <version>2.4.1</version>
                      <configuration>
                          <!-- get all project dependencies -->
                          <descriptorRefs>
                              <descriptorRef>jar-with-dependencies</descriptorRef>
                          </descriptorRefs>
                          <!-- MainClass in mainfest make a executable jar -->
                          <archive>
                              <manifest>
                                  <mainClass>com.alibaba.sgri.message.test</mainClass>
                              </manifest>
                          </archive>
                      </configuration>
                      <executions>
                          <execution>
                              <id>make-assembly</id>
                              <!-- bind to the packaging phase -->
                              <phase>package</phase>
                              <goals>
                                  <goal>single</goal>
                              </goals>
                          </execution>
                      </executions>
                  </plugin>
              </plugins>
          </build>
      </project>

   2. Java プログラムを開発し、JAR パッケージ topn_new.jar を生成します。

      次のコードは Java の例を示しています。

      package com.alibaba.sgri.message;
      import java.io.IOException;
      import java.util.concurrent.atomic.AtomicInteger;
      import com.aliyun.odps.Instance;
      import com.aliyun.odps.Odps;
      import com.aliyun.odps.OdpsException;
      import com.aliyun.odps.account.Account;
      import com.aliyun.odps.account.AliyunAccount;
      import com.aliyun.odps.data.ResultSet;
      import com.aliyun.odps.task.SQLTask;
      import com.dingtalk.api.DefaultDingTalkClient;
      import com.dingtalk.api.DingTalkClient;
      import com.dingtalk.api.request.OapiRobotSendRequest;
      import com.dingtalk.api.response.OapiRobotSendResponse;
      import com.taobao.api.ApiException;
      
      public class test {
      
          public static void main(String[] args) throws ApiException {
              if (args.length < 1) {
                  System.out.println("Please enter the date parameter.");
                  System.exit(0);
              }
              System.out.println("Start reading data.");
              DingTalkClient client = new DefaultDingTalkClient(
                      "<Your chatbot's Webhook URL>");
              OapiRobotSendRequest request = new OapiRobotSendRequest();
              request.setMsgtype("markdown");
              OapiRobotSendRequest.Markdown markdown = new OapiRobotSendRequest.Markdown();
              // ここでは日付がパラメーターとして使用されます。
              markdown.setText(getContent(args[0]));
              markdown.setTitle("Top N jobs by consumption");
              request.setMarkdown(markdown);
              OapiRobotSendResponse response = client.execute(request);
              System.out.println("Message sent successfully.");
          }
      
          /**
           * ODPS から読み取り、送信するデータを取得します。
           */
      
          public static String getContent(String day) {
              Odps odps = createOdps();
              StringBuilder sb = new StringBuilder();
              try {
                  //==================請求アカウント=====================
                  String costTopnSql = "select sum(cost_sum)cost_sum,task_owner from cost_topn where ds='" + day + "' " + "group by task_owner order by cost_sum desc limit 5;";
                  Instance costInstance = SQLTask.run(odps, costTopnSql);
                  costInstance.waitForSuccess();
                  ResultSet costTopnRecords = SQLTask.getResultSet(costInstance);
                  sb.append("<font color=#FF0000 size=4>").append("Top N Billing Accounts (").append(day).append(
                          ")[Calculated based on Alibaba Cloud pay-as-you-go billing]").append("</font>").append("\n\n");
                  AtomicInteger costIndex = new AtomicInteger(1);
                  costTopnRecords.forEach(item -> {
                      sb.append(costIndex.getAndIncrement()).append(".").append("Account:");
                      sb.append("<font color=#2E64FE>").append(item.getString("task_owner")).append("\n\n").append("</font>");
                      sb.append("  ").append(" ").append("Cost:").append("<font color=#2E64FE>").append(item.get("cost_sum"))
                              .append(" CNY").append(
                              "</font>").append("\n\n")
                              .append("</font>");
                  });
                  //==================時間のかかるジョブ=====================
                  String timeTopnSql = "select * from time_topn where ds='" + day + "' ORDER BY cost_time DESC limit 5;";
                  Instance timeInstance = SQLTask.run(odps, timeTopnSql);
                  timeInstance.waitForSuccess();
                  ResultSet timeTopnRecords = SQLTask.getResultSet(timeInstance);
                  sb.append("<font color=#FF8C00 size=4>").append("Top N Time-consuming Jobs (").append(day).append(")")
                          .append("\n\n").append("</font>");
                  AtomicInteger timeIndex = new AtomicInteger(1);
                  timeTopnRecords.forEach(item -> {
                      sb.append(timeIndex.getAndIncrement()).append(".").append("Job:");
                      sb.append("<font color=#2E64FE>").append(item.getString("inst_id")).append("\n\n").append("</font>");
                      sb.append("   ").append("Account:").append("<font color=#2E64FE>").append(item.getString("task_owner")).append("\n\n").append("</font>");
                      sb.append("   ").append("Duration:").append("<font color=#2E64FE>").append(item.get("cost_time"))
                              .append("s").append(
                              "</font>").append("\n\n");
                  });
              } catch (OdpsException | IOException e) {
                  e.printStackTrace();
              }
              return sb.toString();
          }
      
          /**
           * ODPS オブジェクトを作成します。
           */
          public static Odps createOdps() {
              String project = "<project_name>";
              // Alibaba Cloud アカウントの AccessKey は、すべての API にアクセスする権限を持っています。これは高いセキュリティリスクをもたらします。API 呼び出しや O&M を実行するには、RAM ユーザーを作成して使用することを強く推奨します。RAM ユーザーを作成するには、RAM コンソールにログインします。
      		// この例では、AccessKey ID と AccessKey Secret を環境変数に格納する方法を示しています。必要に応じて、構成ファイルに格納することもできます。
      		// AccessKey ID と AccessKey Secret をコードにハードコーディングしないことを強く推奨します。これにより、キーが漏洩する可能性があります。
      		String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
      		String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
              String endPoint = "http://service.odps.aliyun.com/api";
              Account account = new AliyunAccount(accessId, accessKey);
              Odps odps = new Odps(account);
              odps.setEndpoint(endPoint);
              odps.setDefaultProject(project);
              return odps;
          }
      }

   3. 生成された topn_new.jar パッケージを MaxCompute リソースとしてアップロードします。

      詳細については、「MaxCompute リソースの作成と使用」をご参照ください。

3. dingsend という名前の Shell ノードを作成し、topn_new.jar パッケージを参照し、定期的なスケジュールを構成します。

   詳細については、「Shell ノード」をご参照ください。

   次のコードは、Shell ノードで実行するコマンドの例を示しています。

   java -jar  topn_new.jar  $1

   $1 は DataWorks のスケジューリングパラメーターです。Shell ノードの右側で、[スケジューリング設定] をクリックします。[基本プロパティ] セクションで、[パラメーター] を ${yyyymmdd} に設定します。

ステップ 6：先祖ノードと子孫ノードのスケジューリングプロパティの設定とノードの実行

ビジネスフローパネルで、information_history、topn、および dingsend ノードを接続して依存関係を作成します。各ノードの再実行プロパティと先祖ノードの依存関係を構成します。構成が完了したら、ノードを右クリックして [ノードを実行] を選択します。

依存関係の構成方法の詳細については、「同一サイクルスケジューリングの依存関係の構成」をご参照ください。

先祖ノードと子孫ノードの構成方法の詳細については、「ノードコンテキストの構成」をご参照ください。

参考資料

• MaxCompute の請求書と使用詳細の分析

• 請求書詳細の表示

オンラインサポート

MaxCompute の使用中に質問や提案がある場合は、チケットを送信してテクニカルサポートにご連絡ください。