12

最近我一直在使用 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

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

我希望它会帮助别人。

4

3 回答 3

3

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

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

另外，请注意，此代码监视和记录集群状态更改，此外还诊断集群是否因错误而终止，如果发生则抛出异常。

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));
}
于 2014-12-20T00:46:40.023 回答
1

作业流程完成后，集群停止，HDFS 分区丢失。为了防止数据丢失，请配置作业流的最后一步以将结果存储在 Amazon S3 中。

如果JobFlowInstancesDetail:KeepJobFlowAliveWhenNoSteps参数设置为 TRUE，则作业流程将转换到 WAITING 状态，而不是在步骤完成后关闭。

每个作业流程最多允许 256 个步骤。

如果您的工作很耗时，我建议您定期存储结果。

长话短说：没有办法知道什么时候完成。相反，您需要将数据保存为工作的一部分。

于 2012-07-08T07:37:47.423 回答
1

--wait-for-steps创建作业流程时使用选项。

./elastic-mapreduce --create \
...
 --wait-for-steps \
...
于 2013-03-24T07:04:50.287 回答