GitHub PullRequestReviewCommentEvent at 2014-05-27 22:26:02
Repository: elephant-bird
Actor: julienledem
Payload: {"comment":{"url":"https://api.github.com/repos/kevinweil/elephant-bird/pulls/comments/13106484","id":13106484,"diff_hunk":"@@ -0,0 +1,415 @@\n+package com.twitter.elephantbird.util;\n+\n+import com.twitter.elephantbird.mapreduce.input.combine.CompositeInputSplit;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.fs.FileSystem;\n+import org.apache.hadoop.fs.Path;\n+import org.apache.hadoop.io.Text;\n+import org.apache.hadoop.io.serializer.Deserializer;\n+import org.apache.hadoop.io.serializer.SerializationFactory;\n+import org.apache.hadoop.io.serializer.Serializer;\n+import org.apache.hadoop.mapreduce.InputSplit;\n+\n+import java.io.DataInputStream;\n+import java.io.DataOutputStream;\n+import java.io.IOException;\n+import java.util.*;\n+\n+import org.apache.hadoop.util.ReflectionUtils;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+/**\n+ * This code facilitates combining InputSplits, managing the locality vs\n+ * input size skew tradeoff. This code is adapted and cleaned up from\n+ * Apache Pig. It was copied to avoid having to depend on all of Pig for\n+ * this utility.\n+ *\n+ * @author Jonathan Coveney\n+ */\n+public class SplitUtil {\n+  private static final Logger LOG = LoggerFactory.getLogger(SplitUtil.class);\n+\n+  public static final  String COMBINE_SPLIT_SIZE = \"elephantbird.combine.split.size\";\n+\n+  private static long getCombinedSplitSize(Configuration conf) throws IOException {\n+    long splitSize = conf.getLong(COMBINE_SPLIT_SIZE, -1);\n+    if (splitSize == -1) {\n+      splitSize = FileSystem.get(conf).getDefaultBlockSize(new Path(\".\"));\n+    }\n+    return splitSize;\n+  }\n+\n+  private static class Node {\n+    private long length = 0;\n+    private List<ComparableSplit> splits;\n+    private boolean sorted;\n+\n+    public Node() throws IOException, InterruptedException {\n+      length = 0;\n+      splits = new ArrayList<ComparableSplit>();\n+      sorted = false;\n+    }\n+\n+    public void add(ComparableSplit split) throws IOException, InterruptedException {\n+      splits.add(split);\n+      length++;\n+    }\n+\n+    public void remove(ComparableSplit split) {\n+      if (!sorted) {\n+        sort();\n+      }\n+      int index = Collections.binarySearch(splits, split);\n+      if (index >= 0) {\n+        splits.remove(index);\n+        length--;\n+      }\n+    }\n+\n+    public void sort() {\n+      if (!sorted) {\n+        Collections.sort(splits);\n+        sorted = true;\n+      }\n+    }\n+\n+    public List<ComparableSplit> getSplits() {\n+      return splits;\n+    }\n+\n+    public long getLength() {\n+      return length;\n+    }\n+  }\n+\n+  private static Comparator<Node> nodeComparator = new Comparator<Node>() {\n+    @Override\n+    public int compare(Node o1, Node o2) {\n+      return Long.signum(o1.length - o2.length);\n+    }\n+  };\n+\n+  private static final class ComparableSplit implements Comparable<ComparableSplit> {\n+    private InputSplit rawInputSplit;\n+    private Set<Node> nodes;\n+    // id used as a tie-breaker when two splits are of equal size.\n+    private long id;\n+    public ComparableSplit(InputSplit split, long id) {\n+      rawInputSplit = split;\n+      nodes = new HashSet<Node>();\n+      this.id = id;\n+    }\n+\n+    public void add(Node node) {\n+      nodes.add(node);\n+    }\n+\n+    public void removeFromNodes() {\n+      for (Node node : nodes) {\n+        node.remove(this);\n+      }\n+    }\n+\n+    public InputSplit getSplit() {\n+      return rawInputSplit;\n+    }\n+\n+    @Override\n+    public boolean equals(Object other) {\n+      if (other == null || !(other instanceof ComparableSplit))\n+        return false;\n+      return (compareTo((ComparableSplit) other) == 0);\n+    }\n+\n+    @Override\n+    public int hashCode() {\n+      int hashCode = 31 + rawInputSplit.hashCode();\n+      for (Node node : nodes) {\n+        hashCode = 17 * hashCode + node.hashCode();\n+      }\n+      return hashCode;\n+    }\n+\n+    @Override\n+    public int compareTo(ComparableSplit other) {\n+      try {\n+        int cmp = -Long.signum(rawInputSplit.getLength() - other.rawInputSplit.getLength());\n+        // in descending order\n+        return cmp == 0 ? Long.signum(id - other.id) : cmp;\n+      } catch (IOException e) {\n+        throw new RuntimeException(e);\n+      } catch (InterruptedException e) {\n+        throw new RuntimeException(e);\n+      }\n+    }\n+  }\n+\n+  private static class DummySplit extends InputSplit {\n+    private long length;\n+\n+    @Override\n+    public String[] getLocations() {\n+      return null;\n+    }\n+\n+    @Override\n+    public long getLength() {\n+      return length;\n+    }\n+\n+    public void setLength(long length) {\n+      this.length = length;\n+    }\n+  }\n+\n+  private static void removeSplits(List<ComparableSplit> splits) {\n+    for (ComparableSplit split: splits) {\n+      split.removeFromNodes();\n+    }\n+  }\n+\n+  public static List<List<InputSplit>> getCombinedSplits(","path":"core/src/main/java/com/twitter/elephantbird/util/SplitUtil.java","position":172,"original_position":172,"commit_id":"95db2c3afa2f5061c860803ed35e6e5593fa71fd","original_commit_id":"95db2c3afa2f5061c860803ed35e6e5593fa71fd","user":{"login":"julienledem","id":367841,"avatar_url":"https://avatars.githubusercontent.com/u/367841?","gravatar_id":"e3ea861eb75892d64e62ba3c88bc2e0c","url":"https://api.github.com/users/julienledem","html_url":"https://github.com/julienledem","followers_url":"https://api.github.com/users/julienledem/followers","following_url":"https://api.github.com/users/julienledem/following{/other_user}","gists_url":"https://api.github.com/users/julienledem/gists{/gist_id}","starred_url":"https://api.github.com/users/julienledem/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/julienledem/subscriptions","organizations_url":"https://api.github.com/users/julienledem/orgs","repos_url":"https://api.github.com/users/julienledem/repos","events_url":"https://api.github.com/users/julienledem/events{/privacy}","received_events_url":"https://api.github.com/users/julienledem/received_events","type":"User","site_admin":false},"body":"It does deal with locality. With the caveat that we combine splits that don't have all their replicas together. So their is some information loss which is unavoidable as long as we abstract this logic in an InputFormat.\r\nIdeally MapReduce would support sending more than one split to the same task.","created_at":"2014-05-27T22:26:02Z","updated_at":"2014-05-27T22:26:02Z","html_url":"https://github.com/kevinweil/elephant-bird/pull/398#discussion_r13106484","pull_request_url":"https://api.github.com/repos/kevinweil/elephant-bird/pulls/398","_links":{"self":{"href":"https://api.github.com/repos/kevinweil/elephant-bird/pulls/comments/13106484"},"html":{"href":"https://github.com/kevinweil/elephant-bird/pull/398#discussion_r13106484"},"pull_request":{"href":"https://api.github.com/repos/kevinweil/elephant-bird/pulls/398"}}}}