Example: Get task execution progress

Getting the execution progress of a task involves several key concepts: a MaxCompute instance, a MaxCompute task, a Fuxi job, a Fuxi task (stage), and a Fuxi instance (worker). Their relationships are as follows:

• A MaxCompute instance usually corresponds to a single MaxCompute task.

• A MaxCompute task can consist of one or more Fuxi jobs. If an SQL statement is complex, MaxCompute automatically submits multiple Fuxi jobs to Fuxi.

• A Fuxi job consists of one or more stages. Each stage consists of multiple workers. Workers within the same stage share the same data processing logic.

In the job monitoring UI, the directed acyclic graph (DAG) at the top corresponds to a SQLTask, which is a type of MaxCompute task. In this graph, green boxes such as M1 and M2_1 represent stages, and arrows indicate data flow and byte count. A table of Fuxi jobs in the middle of the UI shows the I/O records, I/O bytes, status, and progress for each stage. At the bottom, a panel displays the runtime details for each worker in a selected stage.

The following example shows how to print the execution progress of each stage.

import java.util.List;
import java.util.concurrent.CompletableFuture;
import com.aliyun.odps.Instance;
import com.aliyun.odps.LogView;
import com.aliyun.odps.Odps;
import com.aliyun.odps.OdpsException;
import com.aliyun.odps.account.AliyunAccount;
import com.aliyun.odps.task.SQLTask;
public class InstanceManange {
    private static final String STAGE_FORMAT = "%-26s %13s  %5s  %9s  %7s  %7s  %6s";
    public static void printStage(List<Instance.StageProgress> stageprogresses) throws OdpsException {
        long startTime = System.currentTimeMillis();
        // List<Instance.StageProgress> stageprogresses = instance.getTaskProgress(taskName);
        // STAGES        STATUS  TOTAL  COMPLETED  RUNNING  PENDING  BACKUP
        String HEADER_FORMAT = "%26s %13s  %5s  %9s  %7s  %7s  %6s";
        System.out.println(String.format(HEADER_FORMAT, "STAGES", "STATUS", "TOTAL", "COMPLETED", "RUNNING", "PENDING", "BACKUP"));
        int sumComplete = 0;
        int sumTotal = 0;
        int stageCompletedSum = 0;
        for (Instance.StageProgress progress : stageprogresses) {
            String name = progress.getName();
            // NOTE: getTotalWorkers() does not include backup workers. In most cases, getBackupWorkers() returns 0.
            int backup = progress.getBackupWorkers();
            int total = progress.getTotalWorkers();
            int all = backup + total;
            int running = progress.getRunningWorkers();
            int completed = progress.getTerminatedWorkers();
            int pending = all - completed - running;
            Instance.StageProgress.Status status = progress.getStatus();
            String statusString = status != null ? status.toString() : "NULL";
            // M1_job_0     RUNNING      7          7        0        0       0       0     0
            String vertexStr = String.format(STAGE_FORMAT, name, statusString, total, completed, running, pending, backup);
            System.out.println(vertexStr);
            sumComplete += completed;
            sumTotal += total;
            // Mark the stage as Completed
            if (status == Instance.StageProgress.Status.TERMINATED) {
                stageCompletedSum += 1;
            }
        }
        String FOOTER_FORMAT = "%-15s  %-4s  %-25s";
        String verticesSummary = String.format("STAGES: %02d/%02d", stageCompletedSum, stageprogresses.size());
        final float progress = (sumTotal == 0) ? 0.0f : (float) sumComplete / (float) sumTotal;
        final int progressPercent = (int) (progress * 100);
        String progressStr = "" + progressPercent + "%";
        float et = (float) (System.currentTimeMillis() - startTime) / (float) 1000;
        String elapsedTime = "ELAPSED TIME: " + et + " s";
        System.out.println(String.format(FOOTER_FORMAT, verticesSummary, progressStr, elapsedTime));
        System.out.println();
    }
    public static void main(String[] args) throws Exception {
        String ak = "";
        String sk = "";
        String endpoint = "";
        String project = "";
        String sql = "";
        Odps odps = new Odps(new AliyunAccount(ak, sk));
        odps.setEndpoint(endpoint);
        odps.setDefaultProject(project);
        String taskName = "TestGetProgressTask";
        Instance instance = SQLTask.run(odps, project, sql, taskName, null, null);
        System.out.println("LogView:\n" + new LogView(odps).generateLogView(instance, 24) + "\n");
        instance.getTaskDetailJson2(taskName);
        CompletableFuture<Void> future = CompletableFuture.runAsync((() -> {
            try {
                while (true) {
                    printStage(instance.getTaskProgress(taskName));
                    Thread.sleep(200);
                    if (instance.isSuccessful()) {
                        // After the task succeeds, the returned List is empty.
                        printStage(instance.getTaskProgress(taskName));
                        return;
                    }
                }
            } catch (OdpsException | InterruptedException e) {
                e.printStackTrace();
                return;
            }
        }));
        future.join();
    }
}

The following is sample output.

STAGES                           STATUS      TOTAL  COMPLETED  RUNNING  PENDING  BACKUP
M1_job_0                         RUNNING      1          0        0        1       0
M2_1_job_0                       WAITING      1          0        0        1       0
STAGES: 00/02     0%    ELAPSED TIME: 20.97 s