最近我一直在使用 Amazon Web Services (AWS)，我注意到没有太多关于这个主题的文档，所以我添加了我的解决方案。

我正在使用 Amazon Elastic MapReduce (Amazon EMR) 编写应用程序。计算结束后，我需要对它们创建的文件执行一些工作，因此我需要知道作业流何时完成其工作。

这是检查工作流程是否完成的方法:

AmazonElasticMapReduce mapReduce = new AmazonElasticMapReduceClient(credentials);

DescribeJobFlowsRequest jobAttributes = new DescribeJobFlowsRequest()
    .withJobFlowStates("COMPLETED");

List<JobFlowDetail> jobs = mapReduce.describeJobFlows(jobAttributes).getJobFlows();
JobFlowDetail detail = jobs.get(0);

detail.getJobFlowId(); //the id of one of the completed jobs

您还可以在 DescribeJobFlowsRequest 中查找特定的作业 ID，然后检查该作业是否已完成或失败。

希望对其他人有帮助。

最佳答案

我也遇到了这个问题，这是我目前想出的解决方案。它并不完美，但希望它会有所帮助。作为引用，我使用的是 Java 1.7 和 AWS Java SDK 版本 1.9.13。

请注意，此代码假定您正在等待集群 终止，而不是严格地说步骤；如果您的集群在完成所有步骤后终止，这没问题，但如果您使用的集群在步骤完成后仍保持 Activity 状态，这对您没有太大帮助。

另请注意，此代码会监视和记录集群状态更改，此外还会诊断集群是否因错误终止并在终止时抛出异常。

private void yourMainMethod() {
    RunJobFlowRequest request = ...;

    try {
        RunJobFlowResult submission = emr.runJobFlow(request);
        String jobFlowId = submission.getJobFlowId();
        log.info("Submitted EMR job as job flow id {}", jobFlowId);

        DescribeClusterResult result = 
            waitForCompletion(emr, jobFlowId, 90, TimeUnit.SECONDS);
        diagnoseClusterResult(result, jobFlowId);
    } finally {
        emr.shutdown();
    }
}

private DescribeClusterResult waitForCompletion(
             AmazonElasticMapReduceClient emr, String jobFlowId,
             long sleepTime, TimeUnit timeUnit)
        throws InterruptedException {
    String state = "STARTING";
    while (true) {
        DescribeClusterResult result = emr.describeCluster(
                new DescribeClusterRequest().withClusterId(jobFlowId)
        );
        ClusterStatus status = result.getCluster().getStatus();
        String newState = status.getState();
        if (!state.equals(newState)) {
            log.info("Cluster id {} switched from {} to {}.  Reason: {}.",
                     jobFlowId, state, newState, status.getStateChangeReason());
            state = newState;
        }

        switch (state) {
            case "TERMINATED":
            case "TERMINATED_WITH_ERRORS":
            case "WAITING":
                return result;
        }

        timeUnit.sleep(sleepTime);
    }
}

private void diagnoseClusterResult(DescribeClusterResult result, String jobFlowId) {
    ClusterStatus status = result.getCluster().getStatus();
    ClusterStateChangeReason reason = status.getStateChangeReason();
    ClusterStateChangeReasonCode code = 
        ClusterStateChangeReasonCode.fromValue(reason.getCode());
    switch (code) {
    case ALL_STEPS_COMPLETED:
        log.info("Completed EMR job {}", jobFlowId);
        break;
    default:
        failEMR(jobFlowId, status);
    }
}

private static void failEMR(String jobFlowId, ClusterStatus status) {
    String msg = "EMR cluster run %s terminated with errors.  ClusterStatus = %s";
    throw new RuntimeException(String.format(msg, jobFlowId, status));
}

关于java - 我如何等待 Java 应用程序中的 Elastic MapReduce 作业流完成？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/10758374/