Changeset 27102

Timestamp:

2013-03-20T13:00:34+13:00 (11 years ago)

Author:

jmt12

Message:

Adding in a progress reporter, as the set timeout to 10 hours trick doesn't want to work on Medusa

File:

• gs2-extensions/parallel-building/trunk/src/src/java/org/nzdl/gsdl/HadoopGreenstoneIngest.java (modified) (6 diffs)

gs2-extensions/parallel-building/trunk/src/src/java/org/nzdl/gsdl/HadoopGreenstoneIngest.java

@@ -13 +13 @@
 import java.lang.ProcessBuilder;
 import java.lang.ProcessBuilder.*;
+import java.lang.Thread;
 import java.net.InetAddress;
 import java.util.Map;
@@ -20 +21 @@
 import org.apache.hadoop.io.*;
 import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mapreduce.Mapper.Context;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
@@ -234 +236 @@
       import_process_builder.redirectErrorStream(true);
       import_process_builder.redirectOutput(Redirect.appendTo(import_process_log));
+
+      // - create progress reporter (so Hadoop doesn't time us out)
+      Thread reporter = new HadoopProgressReporter(context, import_process_log);
+      reporter.start();
+
       // - run process
       Process import_process = import_process_builder.start();
@@ -247 +254 @@
       {
         System.err.println("Error! Import command failed (" + e.toString() + ")");
-        System.exit(0);
-      }
+      }
+
+      // - stop the progress reporter as, one way or another, there will be no
+      //   more progress
+      reporter.interrupt();
+      reporter = null; // force gc
 
       // - write end time to log
@@ -283 +294 @@
     conf.set("hadoopprefix", args[3]);
     conf.set("collection", args[4]);
+    // Set the number of retries to 1 - hopefully one of the following will work
+    conf.setInt("mapred.map.max.attempts", 1); // Old Hadoop
+    conf.setInt("mapreduce.map.maxattempts", 1); // Hadoop 2.0.3-alpha
+    conf.setInt("mapreduce.map.max.attempts", 1); // Solution on Web
     // prevent timeouts
     long milli_seconds = 60*60*1000; // 1 hour
     conf.setLong("mapred.task.timeout", milli_seconds);
-
     Job job = new Job(conf, "hadoopgreenstoneingest");
     job.setJarByClass(HadoopGreenstoneIngest.class);
@@ -319 +333 @@
   /** main(String[]) **/
 }
+
+class HadoopProgressReporter
+extends Thread
+{
+
+  private Context hadoop_process;
+
+  private File log_file;
+
+  HadoopProgressReporter(Context hadoop_process, File log_file)
+  {
+    this.hadoop_process = hadoop_process;
+    this.log_file = log_file;
+  }
+
+  public void run()
+  {
+    try
+    {
+      while (!this.isInterrupted())
+      {
+        sleep(60000); // Wait a minute
+        //FileWriter fw1 = new FileWriter(this.log_file, true);
+        //long time = System.currentTimeMillis()/1000;
+        //fw1.write("[" + time + "] HadoopProgressReporter.progress()\n");
+        //fw1.close();
+        this.hadoop_process.progress(); // Inform Hadoop we are still processing
+      }
+    }
+    catch (InterruptedException iex)
+    {
+      // We've been interrupted: no more progress
+    }
+    catch (Exception ex)
+    {
+      ex.printStackTrace();
+    }
+  }
+}