Ответ 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));
}