Как я могу ждать завершения потока задания Elastic MapReduce в приложении Java?

Недавно я работал с 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, а затем проверить, завершилось ли это задание.

Я надеюсь, что это поможет другим.

Ответ 1

Я также столкнулся с этой проблемой, и вот решение, к которому я придумал сейчас. Это не идеально, но, надеюсь, это будет полезно. Для справки, я использую 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));
}

Ответ 2

Как только поток задания завершится, кластер остановится, и раздел HDFS будет потерян. чтобы предотвратить потерю данных, настройте последний шаг потока задания для хранения результатов в Amazon S3.

Если для параметра JobFlowInstancesDetail: KeepJobFlowAliveWhenNoSteps установлено значение TRUE, поток задания будет переход в состояние WAITING, а не завершение работы после завершения шагов.

В каждом потоке заданий допускается не более 256 шагов.

Если ваша работа отнимает много времени, я рекомендую вам периодически сохранять результаты.

Короче говоря: нет способа узнать, когда это будет сделано. Вместо этого вам нужно сохранить свои данные как часть задания.

Ответ 3

Используйте параметр --wait-for-steps при создании потока задания.

./elastic-mapreduce --create \
...
 --wait-for-steps \
...