tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jeag...@apache.org
Subject [08/28] git commit: TEZ-1607. support mr envs in mrrsleep and testorderedwordcount (bikas)
Date Thu, 02 Oct 2014 20:45:55 GMT
TEZ-1607. support mr envs in mrrsleep and testorderedwordcount (bikas)


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

Branch: refs/heads/TEZ-8
Commit: 09bd44e76c3e4fd10a474197b7db0f0e1c78541b
Parents: 97528e9
Author: Bikas Saha <bikas@apache.org>
Authored: Sun Sep 21 15:33:45 2014 -0700
Committer: Bikas Saha <bikas@apache.org>
Committed: Sun Sep 21 15:33:45 2014 -0700

----------------------------------------------------------------------
 CHANGES.txt                                           |  1 +
 .../apache/tez/mapreduce/examples/MRRSleepJob.java    | 14 ++++++++++----
 .../tez/mapreduce/examples/TestOrderedWordCount.java  |  9 +++++++++
 3 files changed, 20 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/tez/blob/09bd44e7/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index f069959..53409fb 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -20,6 +20,7 @@ ALL CHANGES:
   TEZ-1524. Resolve user group information only if ACLs are enabled.
   TEZ-1581. GroupByOrderByMRRTest no longer functional.
   TEZ-1157. Optimize broadcast shuffle to download data only once per host. 
+  TEZ-1607. support mr envs in mrrsleep and testorderedwordcount
 
 Release 0.5.1: Unreleased
 

http://git-wip-us.apache.org/repos/asf/tez/blob/09bd44e7/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
index de8a1ad..bbb4d64 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/MRRSleepJob.java
@@ -524,16 +524,21 @@ public class MRRSleepJob extends Configured implements Tool {
     UserPayload mapUserPayload = TezUtils.createUserPayloadFromConf(mapStageConf);
     int numTasks = generateSplitsInAM ? -1 : numMapper;
 
+    Map<String, String> mapEnv = Maps.newHashMap();
+    MRHelpers.updateEnvBasedOnMRTaskEnv(mapStageConf, mapEnv, true);
+    Map<String, String> reduceEnv = Maps.newHashMap();
+    MRHelpers.updateEnvBasedOnMRTaskEnv(mapStageConf, reduceEnv, false);
+
     Vertex mapVertex = Vertex.create("map", ProcessorDescriptor.create(
         MapProcessor.class.getName()).setUserPayload(mapUserPayload), numTasks,
         MRHelpers.getResourceForMRMapper(mapStageConf));
     mapVertex.addTaskLocalFiles(commonLocalResources)
         .addDataSource("MRInput", dataSource)
-        .setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRMapper(mapStageConf));
+        .setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRMapper(mapStageConf)).setTaskEnvironment(mapEnv);
     vertices.add(mapVertex);
 
     if (iReduceStagesCount > 0
-        && numIReducer > 0) {
+        && numIReducer > 0) {      
       for (int i = 0; i < iReduceStagesCount; ++i) {
         Configuration iconf =
             intermediateReduceStageConfs[i];
@@ -544,7 +549,7 @@ public class MRRSleepJob extends Configured implements Tool {
             MRHelpers.getResourceForMRReducer(intermediateReduceStageConfs[i]));
         ivertex.addTaskLocalFiles(commonLocalResources)
             .setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRReducer(
-                intermediateReduceStageConfs[i]));
+                intermediateReduceStageConfs[i])).setTaskEnvironment(reduceEnv);
         vertices.add(ivertex);
       }
     }
@@ -558,7 +563,8 @@ public class MRRSleepJob extends Configured implements Tool {
       finalReduceVertex.addTaskLocalFiles(commonLocalResources)
           .addDataSink("MROutput", MROutputLegacy.createConfigBuilder(
               finalReduceConf, NullOutputFormat.class).build())
-          .setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRReducer(finalReduceConf));
+          .setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRReducer(finalReduceConf))
+          .setTaskEnvironment(reduceEnv);
       vertices.add(finalReduceVertex);
     } else {
       // Map only job

http://git-wip-us.apache.org/repos/asf/tez/blob/09bd44e7/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
index a36d1d2..e519085 100644
--- a/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
+++ b/tez-tests/src/main/java/org/apache/tez/mapreduce/examples/TestOrderedWordCount.java
@@ -79,6 +79,7 @@ import org.apache.tez.runtime.library.conf.OrderedPartitionedKVEdgeConfig;
 import org.apache.tez.runtime.library.partitioner.HashPartitioner;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
 
 /**
  * An MRR job built on top of word count to return words sorted by
@@ -198,6 +199,11 @@ public class TestOrderedWordCount extends Configured implements Tool
{
       dsd = MRInputLegacy.createConfigBuilder(mapStageConf, TextInputFormat.class, inputPath).build();
     }
 
+    Map<String, String> mapEnv = Maps.newHashMap();
+    MRHelpers.updateEnvBasedOnMRTaskEnv(mapStageConf, mapEnv, true);
+    Map<String, String> reduceEnv = Maps.newHashMap();
+    MRHelpers.updateEnvBasedOnMRTaskEnv(mapStageConf, reduceEnv, false);
+
     Vertex mapVertex;
     ProcessorDescriptor mapProcessorDescriptor =
         ProcessorDescriptor.create(MapProcessor.class.getName())
@@ -210,6 +216,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
       mapVertex = Vertex.create("initialmap", mapProcessorDescriptor, -1,
           MRHelpers.getResourceForMRMapper(mapStageConf));
       mapVertex.setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRMapper(mapStageConf));
+      mapVertex.setTaskEnvironment(mapEnv);
     }
     mapVertex.addTaskLocalFiles(commonLocalResources)
         .addDataSource("MRInput", dsd);
@@ -232,6 +239,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
       intermediateVertex = Vertex.create("intermediate_reducer", iReduceProcessorDescriptor,
           intermediateNumReduceTasks, MRHelpers.getResourceForMRReducer(iReduceStageConf));
       intermediateVertex.setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRReducer(iReduceStageConf));
+      intermediateVertex.setTaskEnvironment(reduceEnv);
     }
     intermediateVertex.addTaskLocalFiles(commonLocalResources);
     vertices.add(intermediateVertex);
@@ -253,6 +261,7 @@ public class TestOrderedWordCount extends Configured implements Tool {
       finalReduceVertex = Vertex.create("finalreduce", finalReduceProcessorDescriptor, 1,
           MRHelpers.getResourceForMRReducer(finalReduceConf));
       finalReduceVertex.setTaskLaunchCmdOpts(MRHelpers.getJavaOptsForMRReducer(finalReduceConf));
+      finalReduceVertex.setTaskEnvironment(reduceEnv);
     }
     finalReduceVertex.addTaskLocalFiles(commonLocalResources);
     finalReduceVertex.addDataSink("MROutput",


Mime
View raw message