Changeset 27546 for gs2-extensions/parallel-building

Timestamp:

2013-06-05T13:02:06+12:00 (11 years ago)

Author:

jmt12

Message:

Adding the ability for the Hadoop Mapper to determine what CPU number it is running on (at that moment - best I can do) - this is useful for multicore machines as otherwise there is no way to tell the 'worker processes' apart

File:

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

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

@@ -169 +169 @@
       String task_id = conf.get("mapred.task.id");
       task_id = task_id.substring(8); // remove "attempt_" prefix
+      // Programatically rewrite the protocol as appropriate for the given
+      // archives directory
+      file_path = file_path.replace("hdfs://", hdfs_prefix);
       // - create a temporary directory
       File greenstone_tmp_dir = new File("/tmp/greenstone");
@@ -180 +183 @@
       FileWriter fw1 = new FileWriter(import_process_log, true);
       long start_time = System.currentTimeMillis()/1000;
-      fw1.write("[Started:" + start_time + "]\n");
-      fw1.write("[Host:" + InetAddress.getLocalHost().getHostName() + "]\n");
-      fw1.write("[Task:" + task_id + "]\n");
-
-      // Programatically rewrite the protocol as appropriate for the given
-      // archives directory
-      file_path = file_path.replace("hdfs://", hdfs_prefix);
-      fw1.write("[Map:" + file_path + " => " + value + "]\n");
+      StringBuffer header_block = new StringBuffer("[Started:");
+      header_block.append(start_time);
+      header_block.append("]\n[Host:");
+      header_block.append(InetAddress.getLocalHost().getHostName());
+      header_block.append("]\n[CPU:");
+      header_block.append(runCommand("getcpu"));
+      header_block.append("]\n[Task:");
+      header_block.append(task_id);
+      header_block.append("]\n[Map:");
+      header_block.append(file_path);
+      header_block.append("=>");
+      header_block.append(value);
+      header_block.append("]\n");
+      fw1.write(header_block.toString());
+      header_block = null;
 
       // - create a temporary manifest file to process this file. Overwrite any
@@ -338 +348 @@
   }
   /** main(String[]) **/
+
+  /** @function runCommand()
+   *
+   * A convenience method that calls an external command and returns its
+   * standard out as a string. Warning! Not safe if the command could return a
+   * large amount of text in the STDERR stream - may infinitely block.
+   *
+   */
+  public static String runCommand(String command)
+  {
+    StringBuffer result = new StringBuffer();
+    try
+    {
+      Runtime run = Runtime.getRuntime() ;
+      Process pr = run.exec(command) ;
+      pr.waitFor() ;
+      BufferedReader buf = new BufferedReader( new InputStreamReader( pr.getInputStream() ) ) ;
+      String line;
+      while ( ( line = buf.readLine() ) != null )
+      {
+        result.append(line);
+      }
+    }
+    catch (Exception ex)
+    {
+      System.err.println("Error! " + ex.getMessage());
+    }
+    return result.toString();
+  }
+  /** runCommand() **/
 }