私もこの問題に遭遇しました。これが私が今思いついた解決策です。完璧ではありませんが、お役に立てば幸いです。参考までに、私は 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));
}