ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [1/2] ignite git commit: IGNITE-3414: Implemented local reducer distribution logic.
Date Tue, 05 Jul 2016 10:27:52 GMT
Repository: ignite
Updated Branches:
  refs/heads/ignite-3414 c9507ed4d -> 2eae6392c


IGNITE-3414: Implemented local reducer distribution logic.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c7add726
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c7add726
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c7add726

Branch: refs/heads/ignite-3414
Commit: c7add726c8334a4a7ecb79a5c28ecd2af442814a
Parents: c9507ed
Author: vozerov-gridgain <vozerov@gridgain.com>
Authored: Tue Jul 5 13:24:35 2016 +0300
Committer: vozerov-gridgain <vozerov@gridgain.com>
Committed: Tue Jul 5 13:24:35 2016 +0300

----------------------------------------------------------------------
 .../IgniteHadoopWeightedMapReducePlanner.java   | 33 +++++++++++++++++---
 1 file changed, 29 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/c7add726/modules/hadoop/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopWeightedMapReducePlanner.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopWeightedMapReducePlanner.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopWeightedMapReducePlanner.java
index 8a99946..9f85a97 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopWeightedMapReducePlanner.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/mapreduce/IgniteHadoopWeightedMapReducePlanner.java
@@ -387,7 +387,8 @@ public class IgniteHadoopWeightedMapReducePlanner extends HadoopAbstractMapReduc
         }
 
         // Assign the rest reducers.
-        assignRemoteReducers(remaining, res);
+        if (remaining > 0)
+            assignRemoteReducers(remaining, res);
 
         return res;
     }
@@ -397,13 +398,37 @@ public class IgniteHadoopWeightedMapReducePlanner extends HadoopAbstractMapReduc
      *
      * @param split Split.
      * @param cnt Reducer count.
+     * @param top Topology.
+     * @param mappers Mappers.
      * @param resMap Reducers result map.
      * @return Number of assigned reducers.
      */
-    private int assignLocalReducers(HadoopInputSplit split, int cnt, Map<UUID, Integer>
resMap) {
-        // TODO;
+    private int assignLocalReducers(HadoopInputSplit split, int cnt, HadoopMapReducePlanTopology
top, Mappers mappers,
+        Map<UUID, Integer> resMap) {
+        // Dereference node.
+        UUID nodeId = mappers.splitToNode.get(split);
+
+        assert nodeId != null;
+
+        // Dereference group.
+        HadoopMapReducePlanGroup grp = top.groupForId(nodeId);
+
+        assert grp != null;
 
-        return 0;
+        // Assign more reducers to the node until threshold is reached.
+        int res = 0;
+
+        while (grp.weight() < reducerMigrationThresholdWeight) {
+            res++;
+
+            grp.weight(grp.weight() + locReducerWeight);
+        }
+
+        // Update result map.
+        if (res > 0)
+            resMap.put(nodeId, res);
+
+        return res;
     }
 
     /**


Mime
View raw message