hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gunt...@apache.org
Subject svn commit: r1579403 - in /hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez: HashTableLoader.java MapRecordProcessor.java RecordProcessor.java ReduceRecordProcessor.java TezCacheAccess.java TezContext.java TezProcessor.java
Date Wed, 19 Mar 2014 21:14:54 GMT
Author: gunther
Date: Wed Mar 19 21:14:53 2014
New Revision: 1579403

URL: http://svn.apache.org/r1579403
Log:
HIVE-6613: Control when spcific Inputs / Outputs are started (Siddharth Seth via Gunther Hagleitner)

Added:
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezCacheAccess.java
Modified:
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezContext.java
    hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java Wed Mar
19 21:14:53 2014
@@ -36,8 +36,8 @@ import org.apache.hadoop.hive.ql.exec.pe
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
 import org.apache.hadoop.hive.serde2.ByteStream.Output;
-import org.apache.hadoop.hive.serde2.SerDeException;
 import org.apache.hadoop.hive.serde2.lazybinary.LazyBinarySerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.Writable;
 import org.apache.tez.runtime.api.LogicalInput;
@@ -75,6 +75,7 @@ public class HashTableLoader implements 
         HiveConf.ConfVars.HIVEHASHTABLELOADFACTOR);
     boolean useLazyRows = HiveConf.getBoolVar(hconf, HiveConf.ConfVars.HIVEMAPJOINLAZYHASHTABLE);
 
+    TezCacheAccess tezCacheAccess = TezCacheAccess.createInstance(hconf);
     // We only check if we can use optimized keys here; that is ok because we don't
     // create optimized keys in MapJoin if hash map doesn't have optimized keys.
     if (!HiveConf.getBoolVar(hconf, HiveConf.ConfVars.HIVEMAPJOINUSEOPTIMIZEDKEYS)) {
@@ -86,7 +87,8 @@ public class HashTableLoader implements 
         continue;
       }
 
-      LogicalInput input = tezContext.getInput(parentToInput.get(pos));
+      String inputName = parentToInput.get(pos);
+      LogicalInput input = tezContext.getInput(inputName);
 
       try {
         KeyValueReader kvReader = (KeyValueReader) input.getReader();
@@ -119,6 +121,9 @@ public class HashTableLoader implements 
       } catch (Exception e) {
         throw new HiveException(e);
       }
+      // Register that the Input has been cached.
+      tezCacheAccess.registerCachedInput(inputName);
+      LOG.info("Setting Input: " + inputName + " as cached");
     }
     if (lastKey == null) {
       lastKey = new MapJoinKeyObject(); // No rows in tables, the key type doesn't matter.

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java Wed
Mar 19 21:14:53 2014
@@ -35,24 +35,26 @@ import org.apache.hadoop.hive.ql.exec.Op
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.exec.mr.ExecMapper.reportStats;
 import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext;
+import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.TezKVOutputCollector;
 import org.apache.hadoop.hive.ql.exec.vector.VectorMapOperator;
 import org.apache.hadoop.hive.ql.log.PerfLogger;
 import org.apache.hadoop.hive.ql.plan.MapWork;
 import org.apache.hadoop.hive.ql.plan.OperatorDesc;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.tez.mapreduce.input.MRInputLegacy;
 import org.apache.tez.mapreduce.processor.MRTaskReporter;
 import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.LogicalOutput;
+import org.apache.tez.runtime.api.TezProcessorContext;
 import org.apache.tez.runtime.library.api.KeyValueReader;
 
 /**
  * Process input from tez LogicalInput and write output - for a map plan
  * Just pump the records through the query plan.
  */
-public class MapRecordProcessor  extends RecordProcessor{
+public class MapRecordProcessor extends RecordProcessor {
 
 
   private MapOperator mapOp;
@@ -63,18 +65,13 @@ public class MapRecordProcessor  extends
   private MapWork mapWork;
 
   @Override
-  void init(JobConf jconf, MRTaskReporter mrReporter, Map<String, LogicalInput> inputs,
-      Map<String, OutputCollector> outMap){
+  void init(JobConf jconf, TezProcessorContext processorContext, MRTaskReporter mrReporter,
+      Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs) throws
Exception {
     perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
-    super.init(jconf, mrReporter, inputs, outMap);
+    super.init(jconf, processorContext, mrReporter, inputs, outputs);
 
     //Update JobConf using MRInput, info like filename comes via this
-    MRInputLegacy mrInput = getMRInput(inputs);
-    try {
-      mrInput.init();
-    } catch (IOException e) {
-      throw new RuntimeException("Failed while initializing MRInput", e);
-    }
+    MRInputLegacy mrInput = TezProcessor.getMRInput(inputs);
     Configuration updatedConf = mrInput.getConfigUpdates();
     if (updatedConf != null) {
       for (Entry<String, String> entry : updatedConf) {
@@ -82,6 +79,14 @@ public class MapRecordProcessor  extends
       }
     }
 
+    createOutputMap();
+    // Start all the Outputs.
+    for (Entry<String, LogicalOutput> outputEntry : outputs.entrySet()) {
+      l4j.info("Starting Output: " + outputEntry.getKey());
+      ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize();
+      outputEntry.getValue().start();
+    }
+
     ObjectCache cache = ObjectCacheFactory.getCache(jconf);
     try {
 
@@ -143,25 +148,10 @@ public class MapRecordProcessor  extends
     perfLogger.PerfLogEnd(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
   }
 
-  private MRInputLegacy getMRInput(Map<String, LogicalInput> inputs) {
-    //there should be only one MRInput
-    MRInputLegacy theMRInput = null;
-    for(LogicalInput inp : inputs.values()){
-      if(inp instanceof MRInputLegacy){
-        if(theMRInput != null){
-          throw new IllegalArgumentException("Only one MRInput is expected");
-        }
-        //a better logic would be to find the alias
-        theMRInput = (MRInputLegacy)inp;
-      }
-    }
-    return theMRInput;
-  }
-
   @Override
   void run() throws IOException{
 
-    MRInputLegacy in = getMRInput(inputs);
+    MRInputLegacy in = TezProcessor.getMRInput(inputs);
     KeyValueReader reader = in.getReader();
 
     //process records until done

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java Wed Mar
19 21:14:53 2014
@@ -16,20 +16,26 @@
  * limitations under the License.
  */
 package org.apache.hadoop.hive.ql.exec.tez;
-import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.lang.management.MemoryMXBean;
 import java.net.URLClassLoader;
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.TezKVOutputCollector;
 import org.apache.hadoop.hive.ql.log.PerfLogger;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.tez.mapreduce.processor.MRTaskReporter;
 import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.LogicalOutput;
+import org.apache.tez.runtime.api.TezProcessorContext;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
 
 /**
  * Process input from tez LogicalInput and write output
@@ -39,7 +45,9 @@ public abstract class RecordProcessor  {
 
   protected JobConf jconf;
   protected Map<String, LogicalInput> inputs;
+  protected Map<String, LogicalOutput> outputs;
   protected Map<String, OutputCollector> outMap;
+  protected TezProcessorContext processorContext;
 
   public static final Log l4j = LogFactory.getLog(RecordProcessor.class);
 
@@ -54,20 +62,22 @@ public abstract class RecordProcessor  {
   protected PerfLogger perfLogger = PerfLogger.getPerfLogger();
   protected String CLASS_NAME = RecordProcessor.class.getName();
 
-
   /**
    * Common initialization code for RecordProcessors
    * @param jconf
+   * @param processorContext the {@link TezProcessorContext}
    * @param mrReporter
-   * @param inputs
-   * @param out
+   * @param inputs map of Input names to {@link LogicalInput}s
+   * @param outputs map of Output names to {@link LogicalOutput}s
+   * @throws Exception
    */
-  void init(JobConf jconf, MRTaskReporter mrReporter, Map<String, LogicalInput> inputs,
-      Map<String, OutputCollector> outMap){
+  void init(JobConf jconf, TezProcessorContext processorContext, MRTaskReporter mrReporter,
+      Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs) throws
Exception {
     this.jconf = jconf;
     this.reporter = mrReporter;
     this.inputs = inputs;
-    this.outMap = outMap;
+    this.outputs = outputs;
+    this.processorContext = processorContext;
 
     // Allocate the bean at the beginning -
     memoryMXBean = ManagementFactory.getMemoryMXBean();
@@ -92,9 +102,9 @@ public abstract class RecordProcessor  {
 
   /**
    * start processing the inputs and writing output
-   * @throws IOException
+   * @throws Exception
    */
-  abstract void run() throws IOException;
+  abstract void run() throws Exception;
 
 
   abstract void close();
@@ -132,4 +142,12 @@ public abstract class RecordProcessor  {
     return 10 * cntr;
   }
 
+  protected void createOutputMap() {
+    Preconditions.checkState(outMap == null, "Outputs should only be setup once");
+    outMap = Maps.newHashMap();
+    for (Entry<String, LogicalOutput> entry : outputs.entrySet()) {
+      TezKVOutputCollector collector = new TezKVOutputCollector(entry.getValue());
+      outMap.put(entry.getKey(), collector);
+    }
+  }
 }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java Wed
Mar 19 21:14:53 2014
@@ -23,6 +23,7 @@ import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -35,6 +36,7 @@ import org.apache.hadoop.hive.ql.exec.Op
 import org.apache.hadoop.hive.ql.exec.Utilities;
 import org.apache.hadoop.hive.ql.exec.mr.ExecMapper.reportStats;
 import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext;
+import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.TezKVOutputCollector;
 import org.apache.hadoop.hive.ql.exec.tez.tools.InputMerger;
 import org.apache.hadoop.hive.ql.log.PerfLogger;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
@@ -53,9 +55,14 @@ import org.apache.hadoop.mapred.OutputCo
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.tez.mapreduce.processor.MRTaskReporter;
+import org.apache.tez.runtime.api.Input;
 import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.LogicalOutput;
+import org.apache.tez.runtime.api.TezProcessorContext;
 import org.apache.tez.runtime.library.api.KeyValuesReader;
 
+import com.google.common.collect.Lists;
+
 /**
  * Process input from tez LogicalInput and write output - for a map plan
  * Just pump the records through the query plan.
@@ -88,10 +95,10 @@ public class ReduceRecordProcessor  exte
   List<Object> row = new ArrayList<Object>(Utilities.reduceFieldNameList.size());
 
   @Override
-  void init(JobConf jconf, MRTaskReporter mrReporter, Map<String, LogicalInput> inputs,
-      Map<String, OutputCollector> outMap){
+  void init(JobConf jconf, TezProcessorContext processorContext, MRTaskReporter mrReporter,
+      Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs) throws
Exception {
     perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
-    super.init(jconf, mrReporter, inputs, outMap);
+    super.init(jconf, processorContext, mrReporter, inputs, outputs);
 
     ObjectCache cache = ObjectCacheFactory.getCache(jconf);
 
@@ -163,6 +170,7 @@ public class ReduceRecordProcessor  exte
       if (dummyOps != null) {
         children.addAll(dummyOps);
       }
+      createOutputMap();
       OperatorUtils.setChildrenCollector(children, outMap);
 
       reducer.setReporter(reporter);
@@ -182,10 +190,20 @@ public class ReduceRecordProcessor  exte
   }
 
   @Override
-  void run() throws IOException{
+  void run() throws Exception {
     List<LogicalInput> shuffleInputs = getShuffleInputs(inputs);
-    KeyValuesReader kvsReader;
+    if (shuffleInputs != null) {
+      l4j.info("Waiting for ShuffleInputs to become ready");
+      processorContext.waitForAllInputsReady(new ArrayList<Input>(shuffleInputs));
+    }
 
+    for (Entry<String, LogicalOutput> outputEntry : outputs.entrySet()) {
+      l4j.info("Starting Output: " + outputEntry.getKey());
+      ((TezKVOutputCollector) outMap.get(outputEntry.getKey())).initialize();
+      outputEntry.getValue().start();
+    }
+
+    KeyValuesReader kvsReader;
     try {
       if(shuffleInputs.size() == 1){
         //no merging of inputs required

Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezCacheAccess.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezCacheAccess.java?rev=1579403&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezCacheAccess.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezCacheAccess.java Wed Mar
19 21:14:53 2014
@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.ql.exec.tez;
+
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.exec.ObjectCache;
+import org.apache.hadoop.hive.ql.exec.ObjectCacheFactory;
+
+/**
+ * Access to the Object cache from Tez, along with utility methods for accessing specific
Keys.
+ */
+public class TezCacheAccess {
+
+  private TezCacheAccess(ObjectCache cache) {
+    this.cache = cache;
+  }
+
+  private ObjectCache cache;
+
+  public static TezCacheAccess createInstance(Configuration conf) {
+    ObjectCache cache = ObjectCacheFactory.getCache(conf);
+    return new TezCacheAccess(cache);
+  }
+
+  private static final String CACHED_INPUT_KEY = "CACHED_INPUTS";
+  
+  private final ReentrantLock cachedInputLock = new ReentrantLock();
+
+  public boolean isInputCached(String inputName) {
+    this.cachedInputLock.lock();
+    try {
+      @SuppressWarnings("unchecked")
+      Set<String> cachedInputs = (Set<String>) cache.retrieve(CACHED_INPUT_KEY);
+      if (cachedInputs == null) {
+        return false;
+      } else {
+        return cachedInputs.contains(inputName);
+      }
+    } finally {
+      this.cachedInputLock.unlock();
+    }
+  }
+
+  public void registerCachedInput(String inputName) {
+    this.cachedInputLock.lock();
+    try {
+      @SuppressWarnings("unchecked")
+      Set<String> cachedInputs = (Set<String>) cache.retrieve(CACHED_INPUT_KEY);
+      if (cachedInputs == null) {
+        cachedInputs = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
+        cache.cache(CACHED_INPUT_KEY, cachedInputs);
+      }
+      cachedInputs.add(inputName);
+    } finally {
+      this.cachedInputLock.unlock();
+    }
+  }
+
+}

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezContext.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezContext.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezContext.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezContext.java Wed Mar 19 21:14:53
2014
@@ -22,6 +22,7 @@ import java.util.Map;
 import org.apache.hadoop.hive.ql.exec.MapredContext;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.tez.runtime.api.LogicalInput;
+import org.apache.tez.runtime.api.LogicalOutput;
 
 /**
  * TezContext contains additional context only available with Tez
@@ -30,6 +31,8 @@ public class TezContext extends MapredCo
 
   // all the inputs for the tez processor
   private Map<String, LogicalInput> inputs;
+  
+  private Map<String, LogicalOutput> outputs;
 
   public TezContext(boolean isMap, JobConf jobConf) {
     super(isMap, jobConf);
@@ -38,6 +41,10 @@ public class TezContext extends MapredCo
   public void setInputs(Map<String, LogicalInput> inputs) {
     this.inputs = inputs;
   }
+  
+  public void setOutputs(Map<String, LogicalOutput> outputs) {
+    this.outputs = outputs;
+  }
 
   public LogicalInput getInput(String name) {
     if (inputs == null) {
@@ -45,4 +52,11 @@ public class TezContext extends MapredCo
     }
     return inputs.get(name);
   }
+  
+  public LogicalOutput getOutput(String name) {
+    if (outputs == null) {
+      return null;
+    }
+    return outputs.get(name);
+  }
 }

Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java?rev=1579403&r1=1579402&r2=1579403&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java Wed Mar 19
21:14:53 2014
@@ -21,15 +21,16 @@ import java.text.NumberFormat;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.ql.exec.tez.TezProcessor.KVOutputCollector;
 import org.apache.hadoop.hive.ql.log.PerfLogger;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.tez.common.TezUtils;
+import org.apache.tez.mapreduce.input.MRInputLegacy;
 import org.apache.tez.mapreduce.processor.MRTaskReporter;
 import org.apache.tez.runtime.api.Event;
 import org.apache.tez.runtime.api.LogicalIOProcessor;
@@ -43,6 +44,9 @@ import org.apache.tez.runtime.library.ap
  * Does what ExecMapper and ExecReducer does for hive in MR framework.
  */
 public class TezProcessor implements LogicalIOProcessor {
+
+  
+  
   private static final Log LOG = LogFactory.getLog(TezProcessor.class);
   private boolean isMap = false;
 
@@ -123,32 +127,34 @@ public class TezProcessor implements Log
     // in case of broadcast-join read the broadcast edge inputs
     // (possibly asynchronously)
 
-    LOG.info("Running map: " + processorContext.getUniqueIdentifier());
-    for (LogicalInput input : inputs.values()) {
-      input.start();
-    }
-    for (LogicalOutput output : outputs.values()) {
-      output.start();
-    }
-
-    Map<String, OutputCollector> outMap = new HashMap<String, OutputCollector>();
+    LOG.info("Running task: " + processorContext.getUniqueIdentifier());
 
-    for (String outputName: outputs.keySet()) {
-      LOG.info("Handling output: " + outputName);
-      KeyValueWriter kvWriter = (KeyValueWriter) outputs.get(outputName).getWriter();
-      OutputCollector collector = new KVOutputCollector(kvWriter);
-      outMap.put(outputName, collector);
-    }
-
-    if(isMap){
+    if (isMap) {
       rproc = new MapRecordProcessor();
-    }
-    else{
+      MRInputLegacy mrInput = getMRInput(inputs);
+      try {
+        mrInput.init();
+      } catch (IOException e) {
+        throw new RuntimeException("Failed while initializing MRInput", e);
+      }
+    } else {
       rproc = new ReduceRecordProcessor();
     }
 
+    TezCacheAccess cacheAccess = TezCacheAccess.createInstance(jobConf);
+    // Start the actual Inputs. After MRInput initialization.
+    for (Entry<String, LogicalInput> inputEntry : inputs.entrySet()) {
+      if (!cacheAccess.isInputCached(inputEntry.getKey())) {
+        inputEntry.getValue().start();
+      } else {
+        LOG.info("Input: " + inputEntry.getKey() + " is already cached. Skipping start");
+      }
+    }
+
+    // Outputs will be started later by the individual Processors.
+
     MRTaskReporter mrReporter = new MRTaskReporter(processorContext);
-    rproc.init(jobConf, mrReporter, inputs, outMap);
+    rproc.init(jobConf, processorContext, mrReporter, inputs, outputs);
     rproc.run();
 
     //done - output does not need to be committed as hive does not use outputcommitter
@@ -156,19 +162,39 @@ public class TezProcessor implements Log
   }
 
   /**
-   * KVOutputCollector. OutputCollector that writes using KVWriter
-   *
+   * KVOutputCollector. OutputCollector that writes using KVWriter.
+   * Must be initialized before it is used.
+   * 
    */
-  static class KVOutputCollector implements OutputCollector {
-    private final KeyValueWriter output;
+  static class TezKVOutputCollector implements OutputCollector {
+    private KeyValueWriter writer;
+    private final LogicalOutput output;
 
-    KVOutputCollector(KeyValueWriter output) {
-      this.output = output;
+    TezKVOutputCollector(LogicalOutput logicalOutput) {
+      this.output = logicalOutput;
+    }
+
+    void initialize() throws Exception {
+      this.writer = (KeyValueWriter) output.getWriter();
     }
 
     public void collect(Object key, Object value) throws IOException {
-        output.write(key, value);
+      writer.write(key, value);
     }
   }
 
+  static  MRInputLegacy getMRInput(Map<String, LogicalInput> inputs) {
+    //there should be only one MRInput
+    MRInputLegacy theMRInput = null;
+    for(LogicalInput inp : inputs.values()){
+      if(inp instanceof MRInputLegacy){
+        if(theMRInput != null){
+          throw new IllegalArgumentException("Only one MRInput is expected");
+        }
+        //a better logic would be to find the alias
+        theMRInput = (MRInputLegacy)inp;
+      }
+    }
+    return theMRInput;
+  }
 }



Mime
View raw message