You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2013/09/14 00:37:00 UTC
git commit: TEZ-420. Change Map/Reduce Processor to work with the new
engine APIs (bikas)
Updated Branches:
refs/heads/TEZ-398 d59d6620c -> ca7abb76f
TEZ-420. Change Map/Reduce Processor to work with the new engine APIs (bikas)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/ca7abb76
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/ca7abb76
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/ca7abb76
Branch: refs/heads/TEZ-398
Commit: ca7abb76f5a6387535c154f8861ab6d072d55900
Parents: d59d662
Author: Bikas Saha <bi...@apache.org>
Authored: Fri Sep 13 15:35:50 2013 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Fri Sep 13 15:35:50 2013 -0700
----------------------------------------------------------------------
.../apache/tez/dag/app/dag/impl/TaskImpl.java | 2 +
.../tez/engine/newapi/LogicalIOProcessor.java | 6 +-
.../engine/lib/input/ShuffledMergedInput.java | 4 +-
.../engine/lib/output/OnFileSortedOutput.java | 3 +-
.../newapi/impl/TezProcessorContextImpl.java | 2 +-
.../mapreduce/hadoop/newmapred/MRReporter.java | 8 +-
.../newmapred/TaskAttemptContextImpl.java | 93 +++
.../hadoop/newmapreduce/MapContextImpl.java | 84 +++
.../newmapreduce/TaskAttemptContextImpl.java | 17 +-
.../TaskInputOutputContextImpl.java | 95 +++
.../newinput/ShuffledMergedInputLegacy.java | 29 +
.../tez/mapreduce/newinput/SimpleInput.java | 20 +-
.../mapreduce/newinput/SimpleInputLegacy.java | 35 ++
.../tez/mapreduce/newoutput/SimpleOutput.java | 3 +-
.../FileSystemStatisticsUpdater.java | 84 +++
.../mapreduce/newprocessor/GcTimeUpdater.java | 71 +++
.../tez/mapreduce/newprocessor/MRTask.java | 597 +++++++++++++++++++
.../mapreduce/newprocessor/MRTaskReporter.java | 101 ++++
.../newprocessor/map/MapProcessor.java | 353 +++++++++++
.../newprocessor/reduce/ReduceProcessor.java | 363 +++++++++++
20 files changed, 1941 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
index b66760c..f0b2ceb 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
@@ -901,6 +901,8 @@ public class TaskImpl implements Task, EventHandler<TaskEvent> {
} else {
// Don't think this can be a pluggable decision, so simply raise an
// event for the TaskAttempt to delete its output.
+ // TODO . Wait for commit attempt to succeed. Dont kill this. If commit
+ // attempt fails then choose a different committer.
LOG.info(task.commitAttempt
+ " already given a go for committing the task output, so killing "
+ attemptID);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-engine-api/src/main/java/org/apache/tez/engine/newapi/LogicalIOProcessor.java
----------------------------------------------------------------------
diff --git a/tez-engine-api/src/main/java/org/apache/tez/engine/newapi/LogicalIOProcessor.java b/tez-engine-api/src/main/java/org/apache/tez/engine/newapi/LogicalIOProcessor.java
index 2ffe060..53b5184 100644
--- a/tez-engine-api/src/main/java/org/apache/tez/engine/newapi/LogicalIOProcessor.java
+++ b/tez-engine-api/src/main/java/org/apache/tez/engine/newapi/LogicalIOProcessor.java
@@ -18,7 +18,6 @@
package org.apache.tez.engine.newapi;
-import java.io.IOException;
import java.util.Map;
/**
@@ -36,10 +35,9 @@ public interface LogicalIOProcessor extends Processor {
* @param outputs
* a map of the destination vertex name to {@link LogicalOutput} -
* one per outgoing edge
- * @throws IOException
- * if an error occurs
+ * @throws Exception TODO
*/
public void run(Map<String, LogicalInput> inputs,
- Map<String, LogicalOutput> outputs) throws IOException;
+ Map<String, LogicalOutput> outputs) throws Exception;
}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-engine/src/main/java/org/apache/tez/engine/lib/input/ShuffledMergedInput.java
----------------------------------------------------------------------
diff --git a/tez-engine/src/main/java/org/apache/tez/engine/lib/input/ShuffledMergedInput.java b/tez-engine/src/main/java/org/apache/tez/engine/lib/input/ShuffledMergedInput.java
index 5d67b0c..da573c5 100644
--- a/tez-engine/src/main/java/org/apache/tez/engine/lib/input/ShuffledMergedInput.java
+++ b/tez-engine/src/main/java/org/apache/tez/engine/lib/input/ShuffledMergedInput.java
@@ -96,7 +96,7 @@ public class ShuffledMergedInput implements LogicalInput {
*/
public void waitForInputReady() throws IOException, InterruptedException {
rawIter = shuffle.waitForInput();
- createValuesIteartor();
+ createValuesIterator();
}
@Override
@@ -153,7 +153,7 @@ public class ShuffledMergedInput implements LogicalInput {
}
@SuppressWarnings({ "rawtypes", "unchecked" })
- private void createValuesIteartor()
+ private void createValuesIterator()
throws IOException {
vIter = new ValuesIterator(rawIter,
(RawComparator) ConfigUtils.getIntermediateInputKeyComparator(conf),
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-engine/src/main/java/org/apache/tez/engine/lib/output/OnFileSortedOutput.java
----------------------------------------------------------------------
diff --git a/tez-engine/src/main/java/org/apache/tez/engine/lib/output/OnFileSortedOutput.java b/tez-engine/src/main/java/org/apache/tez/engine/lib/output/OnFileSortedOutput.java
index 7e57a62..bf112cd 100644
--- a/tez-engine/src/main/java/org/apache/tez/engine/lib/output/OnFileSortedOutput.java
+++ b/tez-engine/src/main/java/org/apache/tez/engine/lib/output/OnFileSortedOutput.java
@@ -33,7 +33,6 @@ import org.apache.tez.engine.newapi.Event;
import org.apache.tez.engine.newapi.KVWriter;
import org.apache.tez.engine.newapi.LogicalOutput;
import org.apache.tez.engine.newapi.TezOutputContext;
-import org.apache.tez.engine.newapi.Writer;
import org.apache.tez.engine.newapi.events.DataMovementEvent;
import org.apache.tez.engine.shuffle.common.ShuffleUtils;
@@ -69,7 +68,7 @@ public class OnFileSortedOutput implements LogicalOutput {
}
@Override
- public Writer getWriter() throws IOException {
+ public KVWriter getWriter() throws IOException {
return new KVWriter() {
@Override
public void write(Object key, Object value) throws IOException {
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezProcessorContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezProcessorContextImpl.java b/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezProcessorContextImpl.java
index f2f830c..e769550 100644
--- a/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezProcessorContextImpl.java
+++ b/tez-engine/src/main/java/org/apache/tez/engine/newapi/impl/TezProcessorContextImpl.java
@@ -72,7 +72,7 @@ public class TezProcessorContextImpl extends TezTaskContextImpl
public void setProgress(float progress) {
runtimeTask.setProgress(progress);
}
-
+
@Override
public void fatalError(Throwable exception, String message) {
super.signalFatalError(exception, message, sourceInfo);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/MRReporter.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/MRReporter.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/MRReporter.java
index 4638ab3..850b768 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/MRReporter.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/MRReporter.java
@@ -3,6 +3,7 @@ package org.apache.tez.mapreduce.hadoop.newmapred;
import org.apache.hadoop.mapred.Counters.Counter;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.Reporter;
+import org.apache.tez.engine.newapi.TezProcessorContext;
import org.apache.tez.engine.newapi.TezTaskContext;
import org.apache.tez.mapreduce.common.Utils;
@@ -10,7 +11,12 @@ public class MRReporter implements Reporter {
private TezTaskContext tezTaskContext;
private InputSplit split;
+ private boolean isProcessorContext = false;
+ public MRReporter(TezProcessorContext tezProcContext) {
+ this(tezProcContext, null);
+ isProcessorContext = true;
+ }
public MRReporter(TezTaskContext tezTaskContext) {
this(tezTaskContext, null);
}
@@ -22,7 +28,7 @@ public class MRReporter implements Reporter {
@Override
public void progress() {
- // Not reporting progress in Tez.
+ //TODO
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/TaskAttemptContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/TaskAttemptContextImpl.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/TaskAttemptContextImpl.java
new file mode 100644
index 0000000..956fcc2
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapred/TaskAttemptContextImpl.java
@@ -0,0 +1,93 @@
+/* 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.tez.mapreduce.hadoop.newmapred;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapreduce.Counter;
+import org.apache.hadoop.util.Progressable;
+import org.apache.tez.mapreduce.newprocessor.MRTaskReporter;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class TaskAttemptContextImpl
+ extends org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
+ implements TaskAttemptContext {
+ private MRTaskReporter reporter;
+
+ // FIXME we need to use DAG Id but we are using App Id
+ public TaskAttemptContextImpl(JobConf conf, TaskAttemptID taskAttemptId,
+ MRTaskReporter reporter) {
+ super(conf, taskAttemptId);
+ this.reporter = reporter;
+ }
+
+ /**
+ * Get the taskAttemptID.
+ *
+ * @return TaskAttemptID
+ */
+ public TaskAttemptID getTaskAttemptID() {
+ return (TaskAttemptID) super.getTaskAttemptID();
+ }
+
+ public Progressable getProgressible() {
+ return reporter;
+ }
+
+ public JobConf getJobConf() {
+ return (JobConf) getConfiguration();
+ }
+
+ @Override
+ public float getProgress() {
+ return reporter.getProgress();
+ }
+
+ @Override
+ public Counter getCounter(Enum<?> counterName) {
+ return (Counter) reporter.getCounter(counterName);
+ }
+
+ @Override
+ public Counter getCounter(String groupName, String counterName) {
+ return (Counter) reporter.getCounter(groupName, counterName);
+ }
+
+ /**
+ * Report progress.
+ */
+ @Override
+ public void progress() {
+ reporter.progress();
+ }
+
+ /**
+ * Set the current status of the task to the given string.
+ */
+ @Override
+ public void setStatus(String status) {
+ setStatusString(status);
+ reporter.setStatus(status);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/MapContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/MapContextImpl.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/MapContextImpl.java
new file mode 100644
index 0000000..39c7194
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/MapContextImpl.java
@@ -0,0 +1,84 @@
+/**
+ * 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.tez.mapreduce.hadoop.newmapreduce;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.MapContext;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.tez.engine.newapi.TezTaskContext;
+
+/**
+ * The context that is given to the {@link Mapper}.
+ * @param <KEYIN> the key input type to the Mapper
+ * @param <VALUEIN> the value input type to the Mapper
+ * @param <KEYOUT> the key output type from the Mapper
+ * @param <VALUEOUT> the value output type from the Mapper
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class MapContextImpl<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
+ extends TaskInputOutputContextImpl<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
+ implements MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
+ private RecordReader<KEYIN,VALUEIN> reader;
+ private InputSplit split;
+
+ public MapContextImpl(Configuration conf, TaskAttemptID taskid,
+ RecordReader<KEYIN,VALUEIN> reader,
+ RecordWriter<KEYOUT,VALUEOUT> writer,
+ OutputCommitter committer,
+ TezTaskContext context,
+ InputSplit split) {
+ super(conf, taskid, writer, committer, context);
+ this.reader = reader;
+ this.split = split;
+ }
+
+ /**
+ * Get the input split for this map.
+ */
+ public InputSplit getInputSplit() {
+ return split;
+ }
+
+ @Override
+ public KEYIN getCurrentKey() throws IOException, InterruptedException {
+ return reader.getCurrentKey();
+ }
+
+ @Override
+ public VALUEIN getCurrentValue() throws IOException, InterruptedException {
+ return reader.getCurrentValue();
+ }
+
+ @Override
+ public boolean nextKeyValue() throws IOException, InterruptedException {
+ return reader.nextKeyValue();
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskAttemptContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskAttemptContextImpl.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskAttemptContextImpl.java
index 67aa628..549841b 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskAttemptContextImpl.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskAttemptContextImpl.java
@@ -20,8 +20,8 @@ package org.apache.tez.mapreduce.hadoop.newmapreduce;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapred.TaskAttemptID;
-import org.apache.hadoop.mapred.TaskID;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskID;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.tez.engine.newapi.TezTaskContext;
@@ -40,16 +40,19 @@ public class TaskAttemptContextImpl
public TaskAttemptContextImpl(Configuration conf,
TezTaskContext taskContext, boolean isMap) {
// TODO NEWTEZ Can the jt Identifier string be taskContext.getUniqueId ?
- super(conf, new TaskAttemptID(
+ this(conf, new TaskAttemptID(
new TaskID(String.valueOf(taskContext.getApplicationId()
.getClusterTimestamp()), taskContext.getApplicationId().getId(),
isMap ? TaskType.MAP : TaskType.REDUCE,
taskContext.getTaskIndex()),
- taskContext.getAttemptNumber()));
- this.taskContext = taskContext;
-
+ taskContext.getAttemptNumber()), taskContext);
}
-
+
+ public TaskAttemptContextImpl(Configuration conf, TaskAttemptID taId, TezTaskContext context) {
+ super(conf, taId);
+ this.taskContext = context;
+ }
+
@Override
public float getProgress() {
// TODO NEWTEZ Will this break anything ?
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskInputOutputContextImpl.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskInputOutputContextImpl.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskInputOutputContextImpl.java
new file mode 100644
index 0000000..a34cf8d
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/newmapreduce/TaskInputOutputContextImpl.java
@@ -0,0 +1,95 @@
+/**
+ * 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.tez.mapreduce.hadoop.newmapreduce;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskInputOutputContext;
+import org.apache.tez.engine.newapi.TezTaskContext;
+
+/**
+ * A context object that allows input and output from the task. It is only
+ * supplied to the {@link Mapper} or {@link Reducer}.
+ * @param <KEYIN> the input key type for the task
+ * @param <VALUEIN> the input value type for the task
+ * @param <KEYOUT> the output key type for the task
+ * @param <VALUEOUT> the output value type for the task
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public abstract class TaskInputOutputContextImpl<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
+ extends TaskAttemptContextImpl
+ implements TaskInputOutputContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
+ private RecordWriter<KEYOUT,VALUEOUT> output;
+ private OutputCommitter committer;
+
+ public TaskInputOutputContextImpl(Configuration conf, TaskAttemptID taskid,
+ RecordWriter<KEYOUT,VALUEOUT> output,
+ OutputCommitter committer,
+ TezTaskContext context) {
+ super(conf, taskid, context);
+ this.output = output;
+ this.committer = committer;
+ }
+
+ /**
+ * Advance to the next key, value pair, returning null if at end.
+ * @return the key object that was read into, or null if no more
+ */
+ public abstract
+ boolean nextKeyValue() throws IOException, InterruptedException;
+
+ /**
+ * Get the current key.
+ * @return the current key object or null if there isn't one
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public abstract
+ KEYIN getCurrentKey() throws IOException, InterruptedException;
+
+ /**
+ * Get the current value.
+ * @return the value object that was read into
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ public abstract VALUEIN getCurrentValue() throws IOException,
+ InterruptedException;
+
+ /**
+ * Generate an output key/value pair.
+ */
+ public void write(KEYOUT key, VALUEOUT value
+ ) throws IOException, InterruptedException {
+ output.write(key, value);
+ }
+
+ public OutputCommitter getOutputCommitter() {
+ return committer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/ShuffledMergedInputLegacy.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/ShuffledMergedInputLegacy.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/ShuffledMergedInputLegacy.java
new file mode 100644
index 0000000..a83620e
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/ShuffledMergedInputLegacy.java
@@ -0,0 +1,29 @@
+/**
+ * <code>ShuffleMergedInput</code> in a {@link LogicalInput} which shuffles
+ * intermediate sorted data, merges them and provides key/<values> to the
+ * consumer.
+ *
+ * The Copy and Merge will be triggered by the initialization - which is handled
+ * by the Tez framework. Input is not consumable until the Copy and Merge are
+ * complete. Methods are provided to check for this, as well as to wait for
+ * completion. Attempting to get a reader on a non-complete input will block.
+ *
+ */
+
+package org.apache.tez.mapreduce.newinput;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.tez.engine.common.sort.impl.TezRawKeyValueIterator;
+import org.apache.tez.engine.lib.input.ShuffledMergedInput;
+
+public class ShuffledMergedInputLegacy extends ShuffledMergedInput {
+
+ @Private
+ public TezRawKeyValueIterator getIterator() throws IOException, InterruptedException {
+ // wait for input so that iterator is available
+ waitForInputReady();
+ return rawIter;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInput.java
index bb39480..8f11739 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInput.java
@@ -80,20 +80,15 @@ public class SimpleInput implements LogicalInput {
private org.apache.hadoop.mapreduce.InputFormat newInputFormat;
@SuppressWarnings("rawtypes")
private org.apache.hadoop.mapreduce.RecordReader newRecordReader;
- private org.apache.hadoop.mapreduce.InputSplit newInputSplit;
+ protected org.apache.hadoop.mapreduce.InputSplit newInputSplit;
@SuppressWarnings("rawtypes")
private InputFormat oldInputFormat;
@SuppressWarnings("rawtypes")
- private RecordReader oldRecordReader;
+ protected RecordReader oldRecordReader;
protected TaskSplitIndex splitMetaInfo = new TaskSplitIndex();
- // Setup the values iterator once, and set value on the same object each time
- // to prevent lots of objects being created.
- private SimpleValueIterator valueIterator = new SimpleValueIterator();
- private SimpleIterable valueIterable = new SimpleIterable(valueIterator);
-
private TezCounter inputRecordCounter;
private TezCounter fileInputByteCounter;
private List<Statistics> fsStats;
@@ -173,6 +168,12 @@ public class SimpleInput implements LogicalInput {
Object key;
Object value;
+ // Setup the values iterator once, and set value on the same object each time
+ // to prevent lots of objects being created.
+ private SimpleValueIterator valueIterator = new SimpleValueIterator();
+ private SimpleIterable valueIterable = new SimpleIterable(valueIterator);
+
+
private final boolean localNewApi = useNewApi;
@SuppressWarnings("unchecked")
@@ -233,6 +234,7 @@ public class SimpleInput implements LogicalInput {
// Not required at the moment. May be required if splits are sent via events.
}
+ @Override
public List<Event> close() throws IOException {
long bytesInPrev = getInputBytes();
if (useNewApi) {
@@ -274,18 +276,16 @@ public class SimpleInput implements LogicalInput {
private static class SimpleValueIterator implements Iterator<Object> {
private Object value;
- int nextCount = 0;
public void setValue(Object value) {
this.value = value;
}
public boolean hasNext() {
- return nextCount == 0;
+ return value != null;
}
public Object next() {
- nextCount++;
Object value = this.value;
this.value = null;
return value;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInputLegacy.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInputLegacy.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInputLegacy.java
new file mode 100644
index 0000000..33cdaab
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newinput/SimpleInputLegacy.java
@@ -0,0 +1,35 @@
+/**
+ * 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.tez.mapreduce.newinput;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.mapred.RecordReader;
+
+public class SimpleInputLegacy extends SimpleInput {
+
+ @Private
+ public org.apache.hadoop.mapreduce.InputSplit getNewInputSplit() {
+ return this.newInputSplit;
+ }
+
+ @SuppressWarnings("rawtypes")
+ public RecordReader getOldRecordReader() {
+ return this.oldRecordReader;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newoutput/SimpleOutput.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newoutput/SimpleOutput.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newoutput/SimpleOutput.java
index 0d774d2..8f67ce0 100644
--- a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newoutput/SimpleOutput.java
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newoutput/SimpleOutput.java
@@ -20,7 +20,6 @@ import org.apache.tez.engine.newapi.Event;
import org.apache.tez.engine.newapi.KVWriter;
import org.apache.tez.engine.newapi.LogicalOutput;
import org.apache.tez.engine.newapi.TezOutputContext;
-import org.apache.tez.engine.newapi.Writer;
import org.apache.tez.mapreduce.common.Utils;
import org.apache.tez.mapreduce.hadoop.MRConfig;
import org.apache.tez.mapreduce.hadoop.newmapred.MRReporter;
@@ -155,7 +154,7 @@ public class SimpleOutput implements LogicalOutput {
}
@Override
- public Writer getWriter() throws IOException {
+ public KVWriter getWriter() throws IOException {
return new KVWriter() {
private final boolean useNewWriter = useNewApi;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/FileSystemStatisticsUpdater.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/FileSystemStatisticsUpdater.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/FileSystemStatisticsUpdater.java
new file mode 100644
index 0000000..5f97049
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/FileSystemStatisticsUpdater.java
@@ -0,0 +1,84 @@
+/**
+ * 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.tez.mapreduce.newprocessor;
+
+import java.util.List;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.tez.common.counters.FileSystemCounter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TezCounters;
+
+ /**
+ * An updater that tracks the last number reported for a given file
+ * system and only creates the counters when they are needed.
+ */
+ class FileSystemStatisticUpdater {
+ private List<FileSystem.Statistics> stats;
+ private TezCounter readBytesCounter, writeBytesCounter,
+ readOpsCounter, largeReadOpsCounter, writeOpsCounter;
+ private String scheme;
+ private TezCounters counters;
+
+ FileSystemStatisticUpdater(TezCounters counters, List<FileSystem.Statistics> stats, String scheme) {
+ this.stats = stats;
+ this.scheme = scheme;
+ this.counters = counters;
+ }
+
+ void updateCounters() {
+ if (readBytesCounter == null) {
+ readBytesCounter = counters.findCounter(scheme,
+ FileSystemCounter.BYTES_READ);
+ }
+ if (writeBytesCounter == null) {
+ writeBytesCounter = counters.findCounter(scheme,
+ FileSystemCounter.BYTES_WRITTEN);
+ }
+ if (readOpsCounter == null) {
+ readOpsCounter = counters.findCounter(scheme,
+ FileSystemCounter.READ_OPS);
+ }
+ if (largeReadOpsCounter == null) {
+ largeReadOpsCounter = counters.findCounter(scheme,
+ FileSystemCounter.LARGE_READ_OPS);
+ }
+ if (writeOpsCounter == null) {
+ writeOpsCounter = counters.findCounter(scheme,
+ FileSystemCounter.WRITE_OPS);
+ }
+ long readBytes = 0;
+ long writeBytes = 0;
+ long readOps = 0;
+ long largeReadOps = 0;
+ long writeOps = 0;
+ for (FileSystem.Statistics stat: stats) {
+ readBytes = readBytes + stat.getBytesRead();
+ writeBytes = writeBytes + stat.getBytesWritten();
+ readOps = readOps + stat.getReadOps();
+ largeReadOps = largeReadOps + stat.getLargeReadOps();
+ writeOps = writeOps + stat.getWriteOps();
+ }
+ readBytesCounter.setValue(readBytes);
+ writeBytesCounter.setValue(writeBytes);
+ readOpsCounter.setValue(readOps);
+ largeReadOpsCounter.setValue(largeReadOps);
+ writeOpsCounter.setValue(writeOps);
+ }
+ }
+
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/GcTimeUpdater.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/GcTimeUpdater.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/GcTimeUpdater.java
new file mode 100644
index 0000000..b05f011
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/GcTimeUpdater.java
@@ -0,0 +1,71 @@
+/**
+ * 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.tez.mapreduce.newprocessor;
+
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.util.List;
+
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.common.counters.TaskCounter;
+
+/**
+ * An updater that tracks the amount of time this task has spent in GC.
+ */
+ class GcTimeUpdater {
+ private long lastGcMillis = 0;
+ private List<GarbageCollectorMXBean> gcBeans = null;
+ TezCounters counters;
+
+ public GcTimeUpdater(TezCounters counters) {
+ this.gcBeans = ManagementFactory.getGarbageCollectorMXBeans();
+ getElapsedGc(); // Initialize 'lastGcMillis' with the current time spent.
+ this.counters = counters;
+ }
+
+ /**
+ * @return the number of milliseconds that the gc has used for CPU
+ * since the last time this method was called.
+ */
+ protected long getElapsedGc() {
+ long thisGcMillis = 0;
+ for (GarbageCollectorMXBean gcBean : gcBeans) {
+ thisGcMillis += gcBean.getCollectionTime();
+ }
+
+ long delta = thisGcMillis - lastGcMillis;
+ this.lastGcMillis = thisGcMillis;
+ return delta;
+ }
+
+ /**
+ * Increment the gc-elapsed-time counter.
+ */
+ public void incrementGcCounter() {
+ if (null == counters) {
+ return; // nothing to do.
+ }
+
+ TezCounter gcCounter =
+ counters.findCounter(TaskCounter.GC_TIME_MILLIS);
+ if (null != gcCounter) {
+ gcCounter.increment(getElapsedGc());
+ }
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTask.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTask.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTask.java
new file mode 100644
index 0000000..a353959
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTask.java
@@ -0,0 +1,597 @@
+/**
+* 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.tez.mapreduce.newprocessor;
+
+import java.io.IOException;
+import java.text.NumberFormat;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.mapred.FileOutputCommitter;
+import org.apache.hadoop.mapred.FileOutputFormat;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.RawKeyValueIterator;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskID;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer;
+import org.apache.hadoop.mapreduce.task.ReduceContextImpl;
+import org.apache.hadoop.util.Progress;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.util.ResourceCalculatorProcessTree;
+import org.apache.tez.common.Constants;
+import org.apache.tez.common.TezTaskStatus.State;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.engine.common.sort.impl.TezRawKeyValueIterator;
+import org.apache.tez.engine.newapi.TezProcessorContext;
+import org.apache.tez.engine.records.OutputContext;
+import org.apache.tez.mapreduce.hadoop.IDConverter;
+import org.apache.tez.mapreduce.hadoop.MRConfig;
+import org.apache.tez.mapreduce.hadoop.newmapred.TaskAttemptContextImpl;
+import org.apache.tez.mapreduce.hadoop.mapreduce.JobContextImpl;
+import org.apache.tez.mapreduce.hadoop.mapreduce.TezNullOutputCommitter;
+//import org.apache.tez.mapreduce.partition.MRPartitioner;
+
+public abstract class MRTask {
+
+ static final Log LOG = LogFactory.getLog(MRTask.class);
+
+ protected JobConf jobConf;
+ protected JobContext jobContext;
+ protected TaskAttemptContext taskAttemptContext;
+ protected OutputCommitter committer;
+
+ // Current counters
+ transient TezCounters counters;
+ protected GcTimeUpdater gcUpdater;
+ private ResourceCalculatorProcessTree pTree;
+ private long initCpuCumulativeTime = 0;
+ protected TezProcessorContext tezEngineTaskContext;
+ protected TaskAttemptID taskAttemptId;
+ protected Progress progress = new Progress();
+
+ boolean isMap;
+
+ /* flag to track whether task is done */
+ AtomicBoolean taskDone = new AtomicBoolean(false);
+
+ /** Construct output file names so that, when an output directory listing is
+ * sorted lexicographically, positions correspond to output partitions.*/
+ private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance();
+ static {
+ NUMBER_FORMAT.setMinimumIntegerDigits(5);
+ NUMBER_FORMAT.setGroupingUsed(false);
+ }
+
+ protected MRTaskReporter mrReporter;
+ protected boolean useNewApi;
+
+ /**
+ * A Map where Key-> URIScheme and value->FileSystemStatisticUpdater
+ */
+ private Map<String, FileSystemStatisticUpdater> statisticUpdaters =
+ new HashMap<String, FileSystemStatisticUpdater>();
+
+ public MRTask(boolean isMap) {
+ this.isMap = isMap;
+ }
+
+ // TODO how to update progress
+ public void initialize(TezProcessorContext context) throws IOException,
+ InterruptedException {
+
+ tezEngineTaskContext = context;
+ counters = context.getCounters();
+ this.taskAttemptId = new TaskAttemptID(
+ new TaskID(
+ Long.toString(context.getApplicationId().getClusterTimestamp()),
+ context.getApplicationId().getId(),
+ (isMap ? TaskType.MAP : TaskType.REDUCE),
+ context.getTaskIndex()),
+ context.getAttemptNumber());
+ // TODO TEZAM4 Figure out initialization / run sequence of Input, Process,
+ // Output. Phase is MR specific.
+ gcUpdater = new GcTimeUpdater(counters);
+
+ byte[] userPayload = context.getUserPayload();
+ Configuration conf = TezUtils.createConfFromUserPayload(userPayload);
+ if (conf instanceof JobConf) {
+ this.jobConf = (JobConf)conf;
+ } else {
+ this.jobConf = new JobConf(conf);
+ }
+ jobConf.set(Constants.TEZ_ENGINE_TASK_ATTEMPT_ID,
+ taskAttemptId.toString());
+
+ initResourceCalculatorPlugin();
+
+ LOG.info("MRTask.inited: taskAttemptId = " + taskAttemptId.toString());
+ }
+
+ private void initResourceCalculatorPlugin() {
+ Class<? extends ResourceCalculatorProcessTree> clazz =
+ this.jobConf.getClass(MRConfig.RESOURCE_CALCULATOR_PROCESS_TREE,
+ null, ResourceCalculatorProcessTree.class);
+ pTree = ResourceCalculatorProcessTree
+ .getResourceCalculatorProcessTree(System.getenv().get("JVM_PID"), clazz, this.jobConf);
+ LOG.info(" Using ResourceCalculatorProcessTree : " + pTree);
+ if (pTree != null) {
+ pTree.updateProcessTree();
+ initCpuCumulativeTime = pTree.getCumulativeCpuTime();
+ }
+ }
+
+ public TezProcessorContext getUmbilical() {
+ return this.tezEngineTaskContext;
+ }
+
+ public void initTask() throws IOException,
+ InterruptedException {
+ this.mrReporter = new MRTaskReporter(tezEngineTaskContext);
+ this.useNewApi = jobConf.getUseNewMapper();
+ TezDAGID dagId = IDConverter.fromMRTaskAttemptId(taskAttemptId).getTaskID()
+ .getVertexID().getDAGId();
+
+ this.jobContext = new JobContextImpl(jobConf, dagId, mrReporter);
+ this.taskAttemptContext =
+ new TaskAttemptContextImpl(jobConf, taskAttemptId, mrReporter);
+
+ if (getState() == State.UNASSIGNED) {
+ setState(State.RUNNING);
+ }
+
+// combineProcessor = null;
+// boolean useCombiner = false;
+// if (useNewApi) {
+// try {
+// useCombiner = (taskAttemptContext.getCombinerClass() != null);
+// } catch (ClassNotFoundException e) {
+// throw new IOException("Could not find combiner class", e);
+// }
+// } else {
+// useCombiner = (job.getCombinerClass() != null);
+// }
+// if (useCombiner) {
+// combineProcessor = new MRCombiner(this);
+// combineProcessor.initialize(job, getTaskReporter());
+// } else {
+// }
+
+ localizeConfiguration(jobConf);
+ }
+
+// public void initPartitioner(JobConf job) throws IOException,
+// InterruptedException {
+// partitioner = new MRPartitioner(this);
+// ((MRPartitioner) partitioner).initialize(job, getTaskReporter());
+// }
+
+ public void initCommitter(JobConf job, boolean useNewApi,
+ boolean useNullCommitter) throws IOException, InterruptedException {
+ if (useNullCommitter) {
+ setCommitter(new TezNullOutputCommitter());
+ return;
+ }
+ if (useNewApi) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("using new api for output committer");
+ }
+ OutputFormat<?, ?> outputFormat = null;
+ try {
+ outputFormat = ReflectionUtils.newInstance(
+ taskAttemptContext.getOutputFormatClass(), job);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException("Unknown OutputFormat", cnfe);
+ }
+ setCommitter(outputFormat.getOutputCommitter(taskAttemptContext));
+ } else {
+ setCommitter(job.getOutputCommitter());
+ }
+
+ Path outputPath = FileOutputFormat.getOutputPath(job);
+ if (outputPath != null) {
+ if ((getCommitter() instanceof FileOutputCommitter)) {
+ FileOutputFormat.setWorkOutputPath(job,
+ ((FileOutputCommitter) getCommitter())
+ .getTaskAttemptPath(taskAttemptContext));
+ } else {
+ FileOutputFormat.setWorkOutputPath(job, outputPath);
+ }
+ }
+ getCommitter().setupTask(taskAttemptContext);
+ }
+
+ public MRTaskReporter getMRReporter() {
+ return mrReporter;
+ }
+
+ public void setState(State state) {
+ // TODO Auto-generated method stub
+
+ }
+
+ public State getState() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ public OutputCommitter getCommitter() {
+ return committer;
+ }
+
+ public void setCommitter(OutputCommitter committer) {
+ this.committer = committer;
+ }
+
+ public TezCounters getCounters() { return counters; }
+
+ public void setConf(JobConf jobConf) {
+ this.jobConf = jobConf;
+ }
+
+ public JobConf getConf() {
+ return this.jobConf;
+ }
+
+ /**
+ * Gets a handle to the Statistics instance based on the scheme associated
+ * with path.
+ *
+ * @param path the path.
+ * @param conf the configuration to extract the scheme from if not part of
+ * the path.
+ * @return a Statistics instance, or null if none is found for the scheme.
+ */
+ @Private
+ public static List<Statistics> getFsStatistics(Path path, Configuration conf) throws IOException {
+ List<Statistics> matchedStats = new ArrayList<FileSystem.Statistics>();
+ path = path.getFileSystem(conf).makeQualified(path);
+ String scheme = path.toUri().getScheme();
+ for (Statistics stats : FileSystem.getAllStatistics()) {
+ if (stats.getScheme().equals(scheme)) {
+ matchedStats.add(stats);
+ }
+ }
+ return matchedStats;
+ }
+
+ @Private
+ public synchronized String getOutputName() {
+ return "part-" + NUMBER_FORMAT.format(taskAttemptId.getTaskID().getId());
+ }
+
+ public void waitBeforeCompletion(MRTaskReporter reporter) throws IOException,
+ InterruptedException {
+ }
+
+ public void outputReady(MRTaskReporter reporter, OutputContext outputContext)
+ throws IOException,
+ InterruptedException {
+ LOG.info("Task: " + taskAttemptId + " reporting outputReady");
+ updateCounters();
+ statusUpdate();
+ }
+
+ public void done() throws IOException, InterruptedException {
+ updateCounters();
+
+ LOG.info("Task:" + taskAttemptId + " is done."
+ + " And is in the process of committing");
+ // TODO change this to use the new context
+ // TODO TEZ Interaciton between Commit and OutputReady. Merge ?
+ if (isCommitRequired()) {
+ // TODO TEZ-439
+// int retries = MAX_RETRIES;
+// setState(TezTaskStatus.State.COMMIT_PENDING);
+// //say the task tracker that task is commit pending
+// // TODO TEZAM2 - Why is the commitRequired check missing ?
+// while (true) {
+// try {
+// umbilical.commitPending(taskAttemptId, status);
+// break;
+// } catch (InterruptedException ie) {
+// // ignore
+// } catch (IOException ie) {
+// LOG.warn("Failure sending commit pending: " +
+// StringUtils.stringifyException(ie));
+// if (--retries == 0) {
+// System.exit(67);
+// }
+// }
+// }
+ //wait for commit approval and commit
+ // TODO EVENTUALLY - Commit is not required for map tasks. skip a couple of RPCs before exiting.
+ commit(committer);
+ }
+ taskDone.set(true);
+ // Make sure we send at least one set of counter increments. It's
+ // ok to call updateCounters() in this thread after comm thread stopped.
+ updateCounters();
+ sendLastUpdate();
+ //signal the tasktracker that we are done
+ //sendDone(umbilical);
+ }
+
+
+ private boolean isCommitRequired() throws IOException {
+ return committer.needsTaskCommit(taskAttemptContext);
+ }
+
+ /**
+ * Send a status update to the task tracker
+ * @param umbilical
+ * @throws IOException
+ */
+ public void statusUpdate() throws IOException, InterruptedException {
+ // TODO call progress update here if not being called within Map/Reduce
+ }
+
+ /**
+ * Sends last status update before sending umbilical.done();
+ */
+ private void sendLastUpdate()
+ throws IOException, InterruptedException {
+ statusUpdate();
+ }
+
+ private void commit(org.apache.hadoop.mapreduce.OutputCommitter committer
+ ) throws IOException {
+ while (!tezEngineTaskContext.canCommit()) {
+ // This will loop till the AM asks for the task to be killed. As
+ // against, the AM sending a signal to the task to kill itself
+ // gracefully.
+ try {
+ Thread.sleep(1000);
+ } catch(InterruptedException ie) {
+ //ignore
+ }
+ }
+
+ // task can Commit now
+ try {
+ LOG.info("Task " + taskAttemptId + " is allowed to commit now");
+ committer.commitTask(taskAttemptContext);
+ return;
+ } catch (IOException iee) {
+ LOG.warn("Failure committing: " +
+ StringUtils.stringifyException(iee));
+ //if it couldn't commit a successfully then delete the output
+ discardOutput(taskAttemptContext);
+ throw iee;
+ }
+ }
+
+ private
+ void discardOutput(TaskAttemptContext taskContext) {
+ try {
+ committer.abortTask(taskContext);
+ } catch (IOException ioe) {
+ LOG.warn("Failure cleaning up: " +
+ StringUtils.stringifyException(ioe));
+ }
+ }
+
+
+ public void updateCounters() {
+ // TODO Auto-generated method stub
+ // TODO TEZAM Implement.
+ Map<String, List<FileSystem.Statistics>> map = new
+ HashMap<String, List<FileSystem.Statistics>>();
+ for(Statistics stat: FileSystem.getAllStatistics()) {
+ String uriScheme = stat.getScheme();
+ if (map.containsKey(uriScheme)) {
+ List<FileSystem.Statistics> list = map.get(uriScheme);
+ list.add(stat);
+ } else {
+ List<FileSystem.Statistics> list = new ArrayList<FileSystem.Statistics>();
+ list.add(stat);
+ map.put(uriScheme, list);
+ }
+ }
+ for (Map.Entry<String, List<FileSystem.Statistics>> entry: map.entrySet()) {
+ FileSystemStatisticUpdater updater = statisticUpdaters.get(entry.getKey());
+ if(updater==null) {//new FileSystem has been found in the cache
+ updater =
+ new FileSystemStatisticUpdater(counters, entry.getValue(),
+ entry.getKey());
+ statisticUpdaters.put(entry.getKey(), updater);
+ }
+ updater.updateCounters();
+ }
+
+ gcUpdater.incrementGcCounter();
+ updateResourceCounters();
+ }
+
+ /**
+ * Updates the {@link TaskCounter#COMMITTED_HEAP_BYTES} counter to reflect the
+ * current total committed heap space usage of this JVM.
+ */
+ private void updateHeapUsageCounter() {
+ long currentHeapUsage = Runtime.getRuntime().totalMemory();
+ counters.findCounter(TaskCounter.COMMITTED_HEAP_BYTES)
+ .setValue(currentHeapUsage);
+ }
+
+ /**
+ * Update resource information counters
+ */
+ void updateResourceCounters() {
+ // Update generic resource counters
+ updateHeapUsageCounter();
+
+ // Updating resources specified in ResourceCalculatorPlugin
+ if (pTree == null) {
+ return;
+ }
+ pTree.updateProcessTree();
+ long cpuTime = pTree.getCumulativeCpuTime();
+ long pMem = pTree.getCumulativeRssmem();
+ long vMem = pTree.getCumulativeVmem();
+ // Remove the CPU time consumed previously by JVM reuse
+ cpuTime -= initCpuCumulativeTime;
+ counters.findCounter(TaskCounter.CPU_MILLISECONDS).setValue(cpuTime);
+ counters.findCounter(TaskCounter.PHYSICAL_MEMORY_BYTES).setValue(pMem);
+ counters.findCounter(TaskCounter.VIRTUAL_MEMORY_BYTES).setValue(vMem);
+ }
+
+
+ public static String normalizeStatus(String status, Configuration conf) {
+ // Check to see if the status string is too long
+ // and truncate it if needed.
+ int progressStatusLength = conf.getInt(
+ MRConfig.PROGRESS_STATUS_LEN_LIMIT_KEY,
+ MRConfig.PROGRESS_STATUS_LEN_LIMIT_DEFAULT);
+ if (status.length() > progressStatusLength) {
+ LOG.warn("Task status: \"" + status + "\" truncated to max limit ("
+ + progressStatusLength + " characters)");
+ status = status.substring(0, progressStatusLength);
+ }
+ return status;
+ }
+
+ protected static <INKEY,INVALUE,OUTKEY,OUTVALUE>
+ org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context
+ createReduceContext(org.apache.hadoop.mapreduce.Reducer
+ <INKEY,INVALUE,OUTKEY,OUTVALUE> reducer,
+ Configuration job,
+ TaskAttemptID taskId,
+ final TezRawKeyValueIterator rIter,
+ org.apache.hadoop.mapreduce.Counter inputKeyCounter,
+ org.apache.hadoop.mapreduce.Counter inputValueCounter,
+ org.apache.hadoop.mapreduce.RecordWriter<OUTKEY,OUTVALUE> output,
+ org.apache.hadoop.mapreduce.OutputCommitter committer,
+ org.apache.hadoop.mapreduce.StatusReporter reporter,
+ RawComparator<INKEY> comparator,
+ Class<INKEY> keyClass, Class<INVALUE> valueClass
+ ) throws IOException, InterruptedException {
+ RawKeyValueIterator r =
+ new RawKeyValueIterator() {
+
+ @Override
+ public boolean next() throws IOException {
+ return rIter.next();
+ }
+
+ @Override
+ public DataInputBuffer getValue() throws IOException {
+ return rIter.getValue();
+ }
+
+ @Override
+ public Progress getProgress() {
+ return rIter.getProgress();
+ }
+
+ @Override
+ public DataInputBuffer getKey() throws IOException {
+ return rIter.getKey();
+ }
+
+ @Override
+ public void close() throws IOException {
+ rIter.close();
+ }
+ };
+ org.apache.hadoop.mapreduce.ReduceContext<INKEY, INVALUE, OUTKEY, OUTVALUE>
+ reduceContext =
+ new ReduceContextImpl<INKEY, INVALUE, OUTKEY, OUTVALUE>(
+ job,
+ taskId,
+ r,
+ inputKeyCounter,
+ inputValueCounter,
+ output,
+ committer,
+ reporter,
+ comparator,
+ keyClass,
+ valueClass);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Using key class: " + keyClass
+ + ", valueClass: " + valueClass);
+ }
+
+ org.apache.hadoop.mapreduce.Reducer<INKEY,INVALUE,OUTKEY,OUTVALUE>.Context
+ reducerContext =
+ new WrappedReducer<INKEY, INVALUE, OUTKEY, OUTVALUE>().getReducerContext(
+ reduceContext);
+
+ return reducerContext;
+ }
+
+ public void taskCleanup()
+ throws IOException, InterruptedException {
+ // set phase for this task
+ statusUpdate();
+ LOG.info("Runnning cleanup for the task");
+ // do the cleanup
+ committer.abortTask(taskAttemptContext);
+ }
+
+ public void localizeConfiguration(JobConf jobConf)
+ throws IOException, InterruptedException {
+ jobConf.set(JobContext.TASK_ID, taskAttemptId.getTaskID().toString());
+ jobConf.set(JobContext.TASK_ATTEMPT_ID, taskAttemptId.toString());
+ jobConf.setInt(JobContext.TASK_PARTITION,
+ taskAttemptId.getTaskID().getId());
+ jobConf.set(JobContext.ID, taskAttemptId.getJobID().toString());
+ }
+
+ public abstract TezCounter getOutputRecordsCounter();
+
+ public abstract TezCounter getInputRecordsCounter();
+
+ public org.apache.hadoop.mapreduce.TaskAttemptContext getTaskAttemptContext() {
+ return taskAttemptContext;
+ }
+
+ public JobContext getJobContext() {
+ return jobContext;
+ }
+
+ public TaskAttemptID getTaskAttemptId() {
+ return taskAttemptId;
+ }
+
+ public TezProcessorContext getTezEngineTaskContext() {
+ return tezEngineTaskContext;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTaskReporter.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTaskReporter.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTaskReporter.java
new file mode 100644
index 0000000..07a7c57
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/MRTaskReporter.java
@@ -0,0 +1,101 @@
+/**
+* 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.tez.mapreduce.newprocessor;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.engine.newapi.TezProcessorContext;
+import org.apache.tez.mapreduce.hadoop.mapred.MRCounters;
+import org.apache.tez.mapreduce.hadoop.newmapred.MRReporter;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class MRTaskReporter
+ extends org.apache.hadoop.mapreduce.StatusReporter
+ implements Reporter {
+
+ private final TezProcessorContext context;
+ private final Reporter reporter;
+
+ private InputSplit split = null;
+
+ public MRTaskReporter(TezProcessorContext context) {
+ this.context = context;
+ this.reporter = new MRReporter(context);
+ }
+
+ public void setProgress(float progress) {
+ context.setProgress(progress);
+ }
+
+ public void setStatus(String status) {
+ reporter.setStatus(status);
+ }
+
+ public float getProgress() {
+ return reporter.getProgress();
+ };
+
+ public void progress() {
+ reporter.progress();
+ }
+
+ public Counters.Counter getCounter(String group, String name) {
+ TezCounter counter = context.getCounters().findCounter(group, name);
+ MRCounters.MRCounter mrCounter = null;
+ if (counter != null) {
+ mrCounter = new MRCounters.MRCounter(counter);
+ }
+ return mrCounter;
+ }
+
+ public Counters.Counter getCounter(Enum<?> name) {
+ TezCounter counter = context.getCounters().findCounter(name);
+ MRCounters.MRCounter mrCounter = null;
+ if (counter != null) {
+ mrCounter = new MRCounters.MRCounter(counter);
+ }
+ return mrCounter;
+ }
+
+ public void incrCounter(Enum<?> key, long amount) {
+ reporter.incrCounter(key, amount);
+ }
+
+ public void incrCounter(String group, String counter, long amount) {
+ reporter.incrCounter(group, counter, amount);
+ }
+
+ public void setInputSplit(InputSplit split) {
+ this.split = split;
+ }
+
+ public InputSplit getInputSplit() throws UnsupportedOperationException {
+ if (split == null) {
+ throw new UnsupportedOperationException("Input only available on map");
+ } else {
+ return split;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/map/MapProcessor.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/map/MapProcessor.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/map/MapProcessor.java
new file mode 100644
index 0000000..ab4fc8e
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/map/MapProcessor.java
@@ -0,0 +1,353 @@
+/**
+ * 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.tez.mapreduce.newprocessor.map;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.MapRunnable;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.lib.map.WrappedMapper;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.dag.api.TezException;
+import org.apache.tez.engine.lib.output.OnFileSortedOutput;
+import org.apache.tez.engine.newapi.Event;
+import org.apache.tez.engine.newapi.KVReader;
+import org.apache.tez.engine.newapi.KVWriter;
+import org.apache.tez.engine.newapi.LogicalIOProcessor;
+import org.apache.tez.engine.newapi.LogicalInput;
+import org.apache.tez.engine.newapi.LogicalOutput;
+import org.apache.tez.engine.newapi.Output;
+import org.apache.tez.engine.newapi.TezProcessorContext;
+import org.apache.tez.mapreduce.hadoop.newmapreduce.MapContextImpl;
+import org.apache.tez.mapreduce.newinput.SimpleInput;
+import org.apache.tez.mapreduce.newinput.SimpleInputLegacy;
+import org.apache.tez.mapreduce.newprocessor.MRTask;
+import org.apache.tez.mapreduce.newprocessor.MRTaskReporter;
+
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class MapProcessor extends MRTask implements LogicalIOProcessor {
+
+ private static final Log LOG = LogFactory.getLog(MapProcessor.class);
+
+ public MapProcessor(){
+ super(true);
+ }
+
+ @Override
+ public void initialize(TezProcessorContext processorContext)
+ throws IOException {
+ try {
+ super.initialize(processorContext);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+
+ @Override
+ public void handleEvents(List<Event> processorEvents) {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void close() throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void run(Map<String, LogicalInput> inputs,
+ Map<String, LogicalOutput> outputs) throws Exception {
+
+ LOG.info("Running map: " + tezEngineTaskContext.getUniqueIdentifier());
+
+ initTask();
+
+ if (inputs.size() != 1
+ || outputs.size() != 1) {
+ throw new IOException("Cannot handle multiple inputs or outputs"
+ + ", inputCount=" + inputs.size()
+ + ", outputCount=" + outputs.size());
+ }
+ LogicalInput in = inputs.values().iterator().next();
+ Output out = outputs.values().iterator().next();
+
+ // Sanity check
+ if (!(in instanceof SimpleInputLegacy)) {
+ throw new IOException(new TezException(
+ "Only Simple Input supported. Input: " + in.getClass()));
+ }
+ SimpleInputLegacy input = (SimpleInputLegacy)in;
+
+ boolean doingShuffle = true;
+ if (!(out instanceof OnFileSortedOutput)) {
+ doingShuffle = false;
+ }
+
+ KVWriter kvWriter = ((OnFileSortedOutput)out).getWriter();
+
+ //TODO fix committer
+// if (out instanceof SimpleOutput) {
+// initCommitter(jobConf, useNewApi, false);
+// ((SimpleOutput)out).setTask(this);
+// } else if (out instanceof SortingOutput) {
+// initCommitter(jobConf, useNewApi, true);
+// initPartitioner(jobConf);
+// ((SortingOutput)out).setTask(this);
+// }
+
+ if (useNewApi) {
+ runNewMapper(jobConf, mrReporter, input, kvWriter, doingShuffle);
+ } else {
+ runOldMapper(jobConf, mrReporter, input, kvWriter, doingShuffle);
+ }
+
+ done();
+ }
+
+ void runOldMapper(
+ final JobConf job,
+ final MRTaskReporter reporter,
+ final SimpleInputLegacy input,
+ final KVWriter output,
+ final boolean doingShuffle
+ ) throws IOException, InterruptedException {
+
+ // Initialize input in-line since it sets parameters which may be used by the processor.
+ // Done only for SimpleInput.
+ // TODO use new method in SimpleInput to get required info
+ //input.initialize(job, master);
+
+ RecordReader in = new OldRecordReader(input);
+
+ OutputCollector collector = new OldOutputCollector(output);
+
+ MapRunnable runner =
+ (MapRunnable)ReflectionUtils.newInstance(job.getMapRunnerClass(), job);
+
+ runner.run(in, collector, (Reporter)reporter);
+ // start the sort phase only if there are reducers
+ this.statusUpdate();
+ }
+
+ private void runNewMapper(final JobConf job,
+ MRTaskReporter reporter,
+ final SimpleInputLegacy in,
+ KVWriter out,
+ final boolean doingShuffle
+ ) throws IOException, InterruptedException {
+
+ // Initialize input in-line since it sets parameters which may be used by the processor.
+ // Done only for SimpleInput.
+ // TODO use new method in SimpleInput to get required info
+ //in.initialize(job, master);
+
+ // make a task context so we can get the classes
+ org.apache.hadoop.mapreduce.TaskAttemptContext taskContext =
+ getTaskAttemptContext();
+
+ // make a mapper
+ org.apache.hadoop.mapreduce.Mapper mapper;
+ try {
+ mapper = (org.apache.hadoop.mapreduce.Mapper)
+ ReflectionUtils.newInstance(taskContext.getMapperClass(), job);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ }
+
+ org.apache.hadoop.mapreduce.RecordReader input =
+ new NewRecordReader(in);
+
+ org.apache.hadoop.mapreduce.RecordWriter output =
+ new NewOutputCollector(out);
+
+ org.apache.hadoop.mapreduce.InputSplit split = in.getNewInputSplit();
+
+ org.apache.hadoop.mapreduce.MapContext
+ mapContext =
+ new MapContextImpl(
+ job, taskAttemptId,
+ input, output,
+ getCommitter(),
+ tezEngineTaskContext, split);
+
+ org.apache.hadoop.mapreduce.Mapper.Context mapperContext =
+ new WrappedMapper().getMapContext(mapContext);
+
+ input.initialize(split, mapperContext);
+ mapper.run(mapperContext);
+ this.statusUpdate();
+ input.close();
+ output.close(mapperContext);
+ }
+
+ private static class NewRecordReader extends
+ org.apache.hadoop.mapreduce.RecordReader {
+ private final SimpleInput in;
+ private KVReader reader;
+
+ private NewRecordReader(SimpleInput in) throws IOException {
+ this.in = in;
+ this.reader = in.getReader();
+ }
+
+ @Override
+ public void initialize(org.apache.hadoop.mapreduce.InputSplit split,
+ TaskAttemptContext context) throws IOException,
+ InterruptedException {
+ //in.initializeNewRecordReader(split, context);
+ }
+
+ @Override
+ public boolean nextKeyValue() throws IOException,
+ InterruptedException {
+ return reader.next();
+ }
+
+ @Override
+ public Object getCurrentKey() throws IOException,
+ InterruptedException {
+ return reader.getCurrentKV().getKey();
+ }
+
+ @Override
+ public Object getCurrentValue() throws IOException,
+ InterruptedException {
+ return reader.getCurrentKV().getValues().iterator().next();
+ }
+
+ @Override
+ public float getProgress() throws IOException, InterruptedException {
+ return in.getProgress();
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+ }
+
+ private static class OldRecordReader implements RecordReader {
+ private final SimpleInputLegacy simpleInput;
+
+ private OldRecordReader(SimpleInputLegacy simpleInput) {
+ this.simpleInput = simpleInput;
+ }
+
+ @Override
+ public boolean next(Object key, Object value) throws IOException {
+ // TODO broken
+// simpleInput.setKey(key);
+// simpleInput.setValue(value);
+// try {
+// return simpleInput.hasNext();
+// } catch (InterruptedException ie) {
+// throw new IOException(ie);
+// }
+ return simpleInput.getOldRecordReader().next(key, value);
+ }
+
+ @Override
+ public Object createKey() {
+ return simpleInput.getOldRecordReader().createKey();
+ }
+
+ @Override
+ public Object createValue() {
+ return simpleInput.getOldRecordReader().createValue();
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return simpleInput.getOldRecordReader().getPos();
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+
+ @Override
+ public float getProgress() throws IOException {
+ try {
+ return simpleInput.getProgress();
+ } catch (InterruptedException ie) {
+ throw new IOException(ie);
+ }
+ }
+ }
+
+ private static class OldOutputCollector
+ implements OutputCollector {
+ private final KVWriter output;
+
+ OldOutputCollector(KVWriter output) {
+ this.output = output;
+ }
+
+ public void collect(Object key, Object value) throws IOException {
+ output.write(key, value);
+ }
+ }
+
+ private class NewOutputCollector
+ extends org.apache.hadoop.mapreduce.RecordWriter {
+ private final KVWriter out;
+
+ NewOutputCollector(KVWriter out) throws IOException {
+ this.out = out;
+ }
+
+ @Override
+ public void write(Object key, Object value) throws IOException, InterruptedException {
+ out.write(key, value);
+ }
+
+ @Override
+ public void close(TaskAttemptContext context
+ ) throws IOException, InterruptedException {
+ }
+ }
+
+ @Override
+ public void localizeConfiguration(JobConf jobConf)
+ throws IOException, InterruptedException {
+ super.localizeConfiguration(jobConf);
+ jobConf.setBoolean(JobContext.TASK_ISMAP, true);
+ }
+
+ @Override
+ public TezCounter getOutputRecordsCounter() {
+ return tezEngineTaskContext.getCounters().findCounter(TaskCounter.MAP_OUTPUT_RECORDS);
+ }
+
+ @Override
+ public TezCounter getInputRecordsCounter() {
+ return tezEngineTaskContext.getCounters().findCounter(TaskCounter.MAP_INPUT_RECORDS);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/ca7abb76/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/reduce/ReduceProcessor.java
----------------------------------------------------------------------
diff --git a/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/reduce/ReduceProcessor.java b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/reduce/ReduceProcessor.java
new file mode 100644
index 0000000..e597979
--- /dev/null
+++ b/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/newprocessor/reduce/ReduceProcessor.java
@@ -0,0 +1,363 @@
+/**
+ * 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.tez.mapreduce.newprocessor.reduce;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.mapred.Counters.Counter;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.util.Progress;
+import org.apache.hadoop.util.Progressable;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.engine.common.ConfigUtils;
+import org.apache.tez.engine.common.sort.impl.TezRawKeyValueIterator;
+import org.apache.tez.engine.lib.output.OnFileSortedOutput;
+import org.apache.tez.engine.newapi.Event;
+import org.apache.tez.engine.newapi.Input;
+import org.apache.tez.engine.newapi.KVReader;
+import org.apache.tez.engine.newapi.KVWriter;
+import org.apache.tez.engine.newapi.LogicalIOProcessor;
+import org.apache.tez.engine.newapi.LogicalInput;
+import org.apache.tez.engine.newapi.LogicalOutput;
+import org.apache.tez.engine.newapi.Output;
+import org.apache.tez.engine.newapi.TezProcessorContext;
+import org.apache.tez.mapreduce.newinput.ShuffledMergedInputLegacy;
+import org.apache.tez.mapreduce.newoutput.SimpleOutput;
+import org.apache.tez.mapreduce.newprocessor.MRTask;
+import org.apache.tez.mapreduce.newprocessor.MRTaskReporter;
+
+
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class ReduceProcessor
+extends MRTask
+implements LogicalIOProcessor {
+
+ private static final Log LOG = LogFactory.getLog(ReduceProcessor.class);
+
+ private Counter reduceInputKeyCounter;
+ private Counter reduceInputValueCounter;
+
+ public ReduceProcessor() {
+ super(false);
+ }
+
+ @Override
+ public void initialize(TezProcessorContext processorContext)
+ throws IOException {
+ try {
+ super.initialize(processorContext);
+ } catch (InterruptedException e) {
+ throw new IOException(e);
+ }
+ }
+
+
+ @Override
+ public void handleEvents(List<Event> processorEvents) {
+ // TODO Auto-generated method stub
+
+ }
+
+ public void close() throws IOException {
+ // TODO Auto-generated method stub
+
+ }
+
+ @Override
+ public void run(Map<String, LogicalInput> inputs,
+ Map<String, LogicalOutput> outputs) throws Exception {
+
+ LOG.info("Running reduce: " + tezEngineTaskContext.getUniqueIdentifier());
+
+ initTask();
+
+ if (outputs.size() <= 0 || outputs.size() > 1) {
+ throw new IOException("Invalid number of outputs"
+ + ", outputCount=" + outputs.size());
+ }
+
+ if (inputs.size() <= 0 || inputs.size() > 1) {
+ throw new IOException("Invalid number of inputs"
+ + ", inputCount=" + inputs.size());
+ }
+
+ Input in = inputs.values().iterator().next();
+ Output out = outputs.values().iterator().next();
+
+// if (out instanceof SimpleOutput) {
+// initCommitter(jobConf, useNewApi, false);
+// } else if (out instanceof SortingOutput) {
+// initCommitter(jobConf, useNewApi, true);
+// initPartitioner(jobConf);
+// ((SortingOutput)out).setTask(this);
+// }
+
+ this.statusUpdate();
+
+ Class keyClass = ConfigUtils.getIntermediateInputKeyClass(jobConf);
+ Class valueClass = ConfigUtils.getIntermediateInputValueClass(jobConf);
+ LOG.info("Using keyClass: " + keyClass);
+ LOG.info("Using valueClass: " + valueClass);
+ RawComparator comparator =
+ ConfigUtils.getInputKeySecondaryGroupingComparator(jobConf);
+ LOG.info("Using comparator: " + comparator);
+
+ reduceInputKeyCounter =
+ mrReporter.getCounter(TaskCounter.REDUCE_INPUT_GROUPS);
+ reduceInputValueCounter =
+ mrReporter.getCounter(TaskCounter.REDUCE_INPUT_RECORDS);
+
+ // Sanity check
+ if (!(in instanceof ShuffledMergedInputLegacy)) {
+ throw new IOException("Illegal input to reduce: " + in.getClass());
+ }
+ ShuffledMergedInputLegacy shuffleInput = (ShuffledMergedInputLegacy)in;
+ KVReader kvReader = shuffleInput.getReader();
+
+ KVWriter kvWriter = null;
+ if((out instanceof SimpleOutput)) {
+ kvWriter = ((SimpleOutput) out).getWriter();
+ } else if ((out instanceof OnFileSortedOutput)) {
+ kvWriter = ((OnFileSortedOutput) out).getWriter();
+ } else {
+ throw new IOException("Illegal input to reduce: " + in.getClass());
+ }
+
+ if (useNewApi) {
+ try {
+ runNewReducer(
+ jobConf,
+ mrReporter,
+ shuffleInput, comparator, keyClass, valueClass,
+ kvWriter);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ }
+ } else {
+ runOldReducer(
+ jobConf, mrReporter,
+ kvReader, comparator, keyClass, valueClass, kvWriter);
+ }
+
+ done();
+ }
+
+ void runOldReducer(JobConf job,
+ final MRTaskReporter reporter,
+ KVReader input,
+ RawComparator comparator,
+ Class keyClass,
+ Class valueClass,
+ final KVWriter output) throws IOException, InterruptedException {
+
+ Reducer reducer =
+ ReflectionUtils.newInstance(job.getReducerClass(), job);
+
+ // make output collector
+
+ OutputCollector collector =
+ new OutputCollector() {
+ public void collect(Object key, Object value)
+ throws IOException {
+ output.write(key, value);
+ }
+ };
+
+ // apply reduce function
+ try {
+ ReduceValuesIterator values =
+ new ReduceValuesIterator(
+ input, reporter, reduceInputValueCounter);
+
+ values.informReduceProgress();
+ while (values.more()) {
+ reduceInputKeyCounter.increment(1);
+ reducer.reduce(values.getKey(), values, collector, reporter);
+ values.informReduceProgress();
+ }
+
+ //Clean up: repeated in catch block below
+ reducer.close();
+ //End of clean up.
+ } catch (IOException ioe) {
+ try {
+ reducer.close();
+ } catch (IOException ignored) {
+ }
+
+ throw ioe;
+ }
+ }
+
+ private static class ReduceValuesIterator<KEY,VALUE>
+ implements Iterator<VALUE> {
+ private Counter reduceInputValueCounter;
+ private KVReader in;
+ private Progressable reporter;
+ private Object currentKey;
+ private Iterator<Object> currentValues;
+
+ public ReduceValuesIterator (KVReader in,
+ Progressable reporter,
+ Counter reduceInputValueCounter)
+ throws IOException {
+ this.reduceInputValueCounter = reduceInputValueCounter;
+ this.in = in;
+ this.reporter = reporter;
+ }
+
+ public boolean more() throws IOException {
+ boolean more = in.next();
+ if(more) {
+ currentKey = in.getCurrentKV().getKey();
+ currentValues = in.getCurrentKV().getValues().iterator();
+ } else {
+ currentKey = null;
+ currentValues = null;
+ }
+ return more;
+ }
+
+ public KEY getKey() throws IOException {
+ return (KEY) currentKey;
+ }
+
+ public void informReduceProgress() {
+ reporter.progress();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return currentValues.hasNext();
+ }
+
+ @Override
+ public VALUE next() {
+ reduceInputValueCounter.increment(1);
+ return (VALUE) currentValues.next();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ }
+
+ void runNewReducer(JobConf job,
+ final MRTaskReporter reporter,
+ ShuffledMergedInputLegacy input,
+ RawComparator comparator,
+ Class keyClass,
+ Class valueClass,
+ final KVWriter out
+ ) throws IOException,InterruptedException,
+ ClassNotFoundException {
+
+ // make a task context so we can get the classes
+ org.apache.hadoop.mapreduce.TaskAttemptContext taskContext = getTaskAttemptContext();
+
+ // make a reducer
+ org.apache.hadoop.mapreduce.Reducer reducer =
+ (org.apache.hadoop.mapreduce.Reducer)
+ ReflectionUtils.newInstance(taskContext.getReducerClass(), job);
+
+ // wrap value iterator to report progress.
+ final TezRawKeyValueIterator rawIter = input.getIterator();
+ TezRawKeyValueIterator rIter = new TezRawKeyValueIterator() {
+ public void close() throws IOException {
+ rawIter.close();
+ }
+ public DataInputBuffer getKey() throws IOException {
+ return rawIter.getKey();
+ }
+ public Progress getProgress() {
+ return rawIter.getProgress();
+ }
+ public DataInputBuffer getValue() throws IOException {
+ return rawIter.getValue();
+ }
+ public boolean next() throws IOException {
+ boolean ret = rawIter.next();
+ reporter.setProgress(rawIter.getProgress().getProgress());
+ return ret;
+ }
+ };
+
+ org.apache.hadoop.mapreduce.RecordWriter trackedRW =
+ new org.apache.hadoop.mapreduce.RecordWriter() {
+
+ @Override
+ public void write(Object key, Object value) throws IOException,
+ InterruptedException {
+ out.write(key, value);
+ }
+
+ @Override
+ public void close(TaskAttemptContext context) throws IOException,
+ InterruptedException {
+ }
+ };
+
+ org.apache.hadoop.mapreduce.Reducer.Context reducerContext =
+ createReduceContext(
+ reducer, job, taskAttemptId,
+ rIter, reduceInputKeyCounter,
+ reduceInputValueCounter,
+ trackedRW,
+ committer,
+ reporter, comparator, keyClass,
+ valueClass);
+
+
+
+ reducer.run(reducerContext);
+ trackedRW.close(reducerContext);
+ }
+
+ @Override
+ public void localizeConfiguration(JobConf jobConf)
+ throws IOException, InterruptedException {
+ super.localizeConfiguration(jobConf);
+ jobConf.setBoolean(JobContext.TASK_ISMAP, false);
+ }
+
+ @Override
+ public TezCounter getOutputRecordsCounter() {
+ return tezEngineTaskContext.getCounters().findCounter(TaskCounter.REDUCE_OUTPUT_RECORDS);
+ }
+
+ @Override
+ public TezCounter getInputRecordsCounter() {
+ return tezEngineTaskContext.getCounters().findCounter(TaskCounter.REDUCE_INPUT_GROUPS);
+ }
+
+}