Updated Branches: refs/heads/master 170ba8eea -> 6f33d586f
CRUNCH-161: Periodically log job progress (similar behavior to hadoop client) Signed-off-by: Josh Wills <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/incubator-crunch/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-crunch/commit/6f33d586 Tree: http://git-wip-us.apache.org/repos/asf/incubator-crunch/tree/6f33d586 Diff: http://git-wip-us.apache.org/repos/asf/incubator-crunch/diff/6f33d586 Branch: refs/heads/master Commit: 6f33d586fc5b8584d84785d8b8521327276af712 Parents: 170ba8e Author: Chao Shi <[email protected]> Authored: Tue Feb 5 10:58:06 2013 +0800 Committer: Josh Wills <[email protected]> Committed: Sun Feb 10 09:20:08 2013 -0800 ---------------------------------------------------------------------- .../org/apache/crunch/impl/mr/exec/CrunchJob.java | 16 +++++++++++++++ .../crunch/impl/mr/run/RuntimeParameters.java | 2 + 2 files changed, 18 insertions(+), 0 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/6f33d586/crunch/src/main/java/org/apache/crunch/impl/mr/exec/CrunchJob.java ---------------------------------------------------------------------- diff --git a/crunch/src/main/java/org/apache/crunch/impl/mr/exec/CrunchJob.java b/crunch/src/main/java/org/apache/crunch/impl/mr/exec/CrunchJob.java index b4981db..f0e5cd1 100644 --- a/crunch/src/main/java/org/apache/crunch/impl/mr/exec/CrunchJob.java +++ b/crunch/src/main/java/org/apache/crunch/impl/mr/exec/CrunchJob.java @@ -27,6 +27,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.crunch.hadoop.mapreduce.lib.jobcontrol.CrunchControlledJob; import org.apache.crunch.impl.mr.plan.MSCROutputHandler; import org.apache.crunch.impl.mr.plan.PlanningParameters; +import org.apache.crunch.impl.mr.run.RuntimeParameters; import org.apache.crunch.io.FileNamingScheme; import org.apache.crunch.io.PathTarget; import org.apache.hadoop.conf.Configuration; @@ -36,6 +37,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.util.StringUtils; +import com.google.common.base.Objects; import com.google.common.collect.Lists; public class CrunchJob extends CrunchControlledJob { @@ -45,6 +47,7 @@ public class CrunchJob extends CrunchControlledJob { private final Path workingPath; private final Map<Integer, PathTarget> multiPaths; private final boolean mapOnlyJob; + private String lastKnownProgress; public CrunchJob(Job job, Path workingPath, MSCROutputHandler handler) throws IOException { super(job, Lists.<CrunchControlledJob> newArrayList()); @@ -117,6 +120,10 @@ public class CrunchJob extends CrunchControlledJob { this.state = State.FAILED; this.message = "Job failed!"; } + } else { // still running + if (job.getConfiguration().getBoolean(RuntimeParameters.LOG_JOB_PROGRESS, false)) { + logJobProgress(); + } } } catch (IOException ioe) { this.state = State.FAILED; @@ -142,6 +149,15 @@ public class CrunchJob extends CrunchControlledJob { } } + private void logJobProgress() throws IOException, InterruptedException { + String progress = String.format("map %.0f%% reduce %.0f%%", + 100.0 * job.mapProgress(), 100.0 * job.reduceProgress()); + if (!Objects.equal(lastKnownProgress, progress)) { + log.info(job.getJobName() + " progress: " + progress); + lastKnownProgress = progress; + } + } + /** * Extract the partition number from a raw reducer output filename. * http://git-wip-us.apache.org/repos/asf/incubator-crunch/blob/6f33d586/crunch/src/main/java/org/apache/crunch/impl/mr/run/RuntimeParameters.java ---------------------------------------------------------------------- diff --git a/crunch/src/main/java/org/apache/crunch/impl/mr/run/RuntimeParameters.java b/crunch/src/main/java/org/apache/crunch/impl/mr/run/RuntimeParameters.java index 27a8402..1ee19e7 100644 --- a/crunch/src/main/java/org/apache/crunch/impl/mr/run/RuntimeParameters.java +++ b/crunch/src/main/java/org/apache/crunch/impl/mr/run/RuntimeParameters.java @@ -28,6 +28,8 @@ public class RuntimeParameters { public static final String TMP_DIR = "crunch.tmp.dir"; + public static final String LOG_JOB_PROGRESS = "crunch.log.job.progress"; + // Not instantiated private RuntimeParameters() { }
