Changeset 28312 for gs2-extensions/parallel-building/trunk/src/src/java/org/nzdl/gsdl/HadoopGreenstoneIngest2.java

Timestamp:

2013-09-26T11:13:14+12:00 (11 years ago)

Author:

jmt12

Message:

Working on finer control over data locality - so I can configure a run with no locality, for example

File:

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

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

@@ -13 +13 @@
 import java.net.InetAddress;
 import java.nio.channels.FileChannel;
+import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
+import java.util.StringTokenizer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -21 +24 @@
 import org.apache.hadoop.conf.*;
 import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapreduce.*;
 import org.apache.hadoop.mapreduce.Mapper.Context;
@@ -127 +133 @@
     extends FileInputFormat<Text, Text>
   {
+
+    private String[] getActiveServersList(JobContext context)
+    {
+      String [] servers = null;
+      try
+      {
+        JobClient jc = new JobClient((JobConf)context.getConfiguration());
+        ClusterStatus status = jc.getClusterStatus(true);
+        Collection<String> atc = status.getActiveTrackerNames();
+        servers = new String[atc.size()];
+        int s = 0;
+        for (String serverInfo : atc)
+        {
+          StringTokenizer st = new StringTokenizer(serverInfo, ":");
+          String trackerName = st.nextToken();
+          StringTokenizer st1 = new StringTokenizer(trackerName, "_");
+          st1.nextToken();
+          servers[s++] = st1.nextToken();
+        }
+      }
+      catch (IOException e)
+      {
+        e.printStackTrace();
+      }
+      System.err.print("Servers: ");
+      String sep = "";
+      for (Object obj : servers)
+      {
+        System.err.print(sep + obj.toString());
+        sep = ", ";
+      }
+      System.err.println("");
+      return servers;
+    }
+    /** getActiveServersList() **/
+
+    /**
+     */
+    public List<InputSplit> getSplits(JobContext job)
+      throws IOException
+    {
+      System.err.println("GSFileInputFormat::getSplits()");
+      // get splits
+      List<InputSplit> original_splits = super.getSplits(job);
+      // Get active servers
+      String[] servers = getActiveServersList(job);
+      if(servers == null)
+      {
+        return null;
+      }
+      // done
+      System.err.println("Splits: ");
+      for (InputSplit obj : original_splits)
+      {
+        System.err.println(obj.toString());
+      }
+      return original_splits;
+    }
+    /** getSplits() **/
+
     /**
      *  Don't split the files