You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ac...@apache.org on 2013/03/15 22:26:48 UTC
svn commit: r1457129 [15/38] - in /incubator/tez: ./ tez-ampool/
tez-ampool/src/ tez-ampool/src/main/ tez-ampool/src/main/bin/
tez-ampool/src/main/conf/ tez-ampool/src/main/java/
tez-ampool/src/main/java/org/ tez-ampool/src/main/java/org/apache/ tez-am...
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/SimpleInput.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/SimpleInput.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/SimpleInput.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/input/SimpleInput.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,369 @@
+/**
+ * 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.input;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.serializer.Deserializer;
+import org.apache.hadoop.io.serializer.SerializationFactory;
+import org.apache.hadoop.mapred.FileSplit;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitIndex;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.api.Input;
+import org.apache.tez.api.Master;
+import org.apache.tez.common.TezTask;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.mapreduce.processor.MRTask;
+import org.apache.tez.mapreduce.processor.MRTaskReporter;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+
+/**
+ * {@link SimpleInput} is an {@link Input} which provides key/values pairs
+ * for the consumer.
+ *
+ * It is compatible with all standard Apache Hadoop MapReduce
+ * {@link InputFormat} implementations.
+ */
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class SimpleInput implements Input {
+
+ private static final Log LOG = LogFactory.getLog(SimpleInput.class);
+
+ MRTask task;
+
+ boolean useNewApi;
+
+ JobConf jobConf;
+
+ org.apache.hadoop.mapreduce.TaskAttemptContext taskAttemptContext;
+
+ org.apache.hadoop.mapreduce.InputFormat newInputFormat;
+ org.apache.hadoop.mapreduce.RecordReader newRecordReader;
+
+ org.apache.hadoop.mapred.InputFormat oldInputFormat;
+ org.apache.hadoop.mapred.RecordReader oldRecordReader;
+
+ Object key;
+ Object value;
+
+ private TezCounter inputRecordCounter;
+ private TezCounter fileInputByteCounter;
+ private List<Statistics> fsStats;
+ private MRTaskReporter reporter;
+
+ @Inject
+ public SimpleInput(
+ @Assisted TezTask task
+ )
+ {}
+
+ public void setTask(MRTask task) {
+ this.task = task;
+ }
+
+ public void initialize(Configuration conf, Master master) throws IOException,
+ InterruptedException {
+ if (task == null) {
+ return;
+ }
+
+ if (conf instanceof JobConf) {
+ jobConf = (JobConf)conf;
+ } else {
+ jobConf = new JobConf(conf);
+ }
+
+ useNewApi = jobConf.getUseNewMapper();
+ taskAttemptContext = task.getTaskAttemptContext();
+
+ inputRecordCounter = task.getInputRecordsCounter();
+ fileInputByteCounter = task.getFileInputBytesCounter();
+
+ reporter = task.getMRReporter();
+
+ if (useNewApi) {
+ try {
+ newInputFormat =
+ ReflectionUtils.newInstance(
+ taskAttemptContext.getInputFormatClass(), jobConf);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ }
+
+ newInputSplit = getNewSplitDetails(task.getSplitIndex());
+ List<Statistics> matchedStats = null;
+ if (newInputSplit instanceof org.apache.hadoop.mapreduce.lib.input.FileSplit) {
+ matchedStats = MRTask.getFsStatistics(
+ ((org.apache.hadoop.mapreduce.lib.input.FileSplit)
+ newInputSplit).getPath(), jobConf);
+ }
+ fsStats = matchedStats;
+ newRecordReader =
+ newInputFormat.createRecordReader(newInputSplit, taskAttemptContext);
+ } else {
+ oldInputFormat = jobConf.getInputFormat();
+ org.apache.hadoop.mapred.InputSplit oldInputSplit =
+ getOldSplitDetails(task.getSplitIndex());
+
+ List<Statistics> matchedStats = null;
+ if (oldInputSplit instanceof FileSplit) {
+ matchedStats =
+ MRTask.getFsStatistics(
+ ((FileSplit)oldInputSplit).getPath(), jobConf);
+ }
+ fsStats = matchedStats;
+
+ long bytesInPrev = getInputBytes();
+ oldRecordReader =
+ jobConf.getInputFormat().getRecordReader(
+ oldInputSplit, jobConf, reporter);
+ long bytesInCurr = getInputBytes();
+ fileInputByteCounter.increment(bytesInCurr - bytesInPrev);
+
+ updateJobWithSplit(jobConf, oldInputSplit);
+ }
+ }
+
+ public boolean hasNext() throws IOException, InterruptedException {
+ boolean hasNext = false;
+ long bytesInPrev = getInputBytes();
+
+ if (useNewApi) {
+ hasNext = newRecordReader.nextKeyValue();
+ } else {
+ hasNext = oldRecordReader.next(key, value);
+ }
+
+ long bytesInCurr = getInputBytes();
+ fileInputByteCounter.increment(bytesInCurr - bytesInPrev);
+ reporter.setProgress(getProgress());
+
+ if (hasNext) {
+ inputRecordCounter.increment(1);
+ }
+
+ return hasNext;
+ }
+
+ private SimpleValueIterator vIter = new SimpleValueIterator();
+ private SimpleIterable valuesIterable =
+ new SimpleIterable(vIter);
+
+ private org.apache.hadoop.mapreduce.InputSplit newInputSplit;
+
+ public void setKey(Object key) {
+ this.key = key;
+ }
+
+ public void setValue(Object value) {
+ this.value = value;
+ }
+
+ public Object getNextKey() throws IOException, InterruptedException {
+ if (useNewApi) {
+ return newRecordReader.getCurrentKey();
+ } else {
+ return key;
+ }
+ }
+
+ public Iterable getNextValues() throws IOException,
+ InterruptedException {
+ value = newRecordReader.getCurrentValue();
+ vIter.setValue(value);
+ return valuesIterable;
+ }
+
+ public float getProgress() throws IOException, InterruptedException {
+ if (useNewApi) {
+ return newRecordReader.getProgress();
+ } else {
+ return oldRecordReader.getProgress();
+ }
+ }
+
+ public void close() throws IOException {
+ long bytesInPrev = getInputBytes();
+ if (useNewApi) {
+ newRecordReader.close();
+ } else {
+ oldRecordReader.close();
+ }
+ long bytesInCurr = getInputBytes();
+ fileInputByteCounter.increment(bytesInCurr - bytesInPrev);
+ }
+
+ static class SimpleValueIterator implements Iterator {
+
+ private Object value;
+
+ public void setValue(Object value) {
+ this.value = value;
+ }
+
+ public boolean hasNext() {
+ return false;
+ }
+
+ public Object next() {
+ Object value = this.value;
+ this.value = null;
+ return value;
+ }
+
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ static class SimpleIterable implements Iterable {
+ private final Iterator iterator;
+ public SimpleIterable(Iterator iterator) {
+ this.iterator = iterator;
+ }
+
+ public Iterator iterator() {
+ return iterator;
+ }
+ }
+
+
+ public RecordReader getOldRecordReader() {
+ return oldRecordReader;
+ }
+
+ public org.apache.hadoop.mapreduce.RecordReader getNewRecordReader() {
+ return newRecordReader;
+ }
+
+ public org.apache.hadoop.mapred.InputSplit
+ getOldSplitDetails(TaskSplitIndex splitMetaInfo)
+ throws IOException {
+ Path file = new Path(splitMetaInfo.getSplitLocation());
+ long offset = splitMetaInfo.getStartOffset();
+
+ FileSystem fs = file.getFileSystem(jobConf);
+ FSDataInputStream inFile = fs.open(file);
+ inFile.seek(offset);
+ String className = Text.readString(inFile);
+ Class<org.apache.hadoop.mapred.InputSplit> cls;
+ try {
+ cls =
+ (Class<org.apache.hadoop.mapred.InputSplit>)
+ jobConf.getClassByName(className);
+ } catch (ClassNotFoundException ce) {
+ IOException wrap = new IOException("Split class " + className +
+ " not found");
+ wrap.initCause(ce);
+ throw wrap;
+ }
+ SerializationFactory factory = new SerializationFactory(jobConf);
+ Deserializer<org.apache.hadoop.mapred.InputSplit> deserializer =
+ (Deserializer<org.apache.hadoop.mapred.InputSplit>)
+ factory.getDeserializer(cls);
+ deserializer.open(inFile);
+ org.apache.hadoop.mapred.InputSplit split = deserializer.deserialize(null);
+ long pos = inFile.getPos();
+ reporter.getCounter(TaskCounter.SPLIT_RAW_BYTES).increment(pos - offset);
+ inFile.close();
+ return split;
+ }
+
+ public org.apache.hadoop.mapreduce.InputSplit
+ getNewSplitDetails(TaskSplitIndex splitMetaInfo)
+ throws IOException {
+ Path file = new Path(splitMetaInfo.getSplitLocation());
+ long offset = splitMetaInfo.getStartOffset();
+
+ FileSystem fs = file.getFileSystem(jobConf);
+ FSDataInputStream inFile = fs.open(file);
+ inFile.seek(offset);
+ String className = Text.readString(inFile);
+ Class<org.apache.hadoop.mapreduce.InputSplit> cls;
+ try {
+ cls =
+ (Class<org.apache.hadoop.mapreduce.InputSplit>)
+ jobConf.getClassByName(className);
+ } catch (ClassNotFoundException ce) {
+ IOException wrap = new IOException("Split class " + className +
+ " not found");
+ wrap.initCause(ce);
+ throw wrap;
+ }
+ SerializationFactory factory = new SerializationFactory(jobConf);
+ Deserializer<org.apache.hadoop.mapreduce.InputSplit> deserializer =
+ (Deserializer<org.apache.hadoop.mapreduce.InputSplit>)
+ factory.getDeserializer(cls);
+ deserializer.open(inFile);
+ org.apache.hadoop.mapreduce.InputSplit split =
+ deserializer.deserialize(null);
+ long pos = inFile.getPos();
+ reporter.getCounter(TaskCounter.SPLIT_RAW_BYTES).increment(pos - offset);
+ inFile.close();
+ return split;
+ }
+
+ private void updateJobWithSplit(final JobConf job, InputSplit inputSplit) {
+ if (inputSplit instanceof FileSplit) {
+ FileSplit fileSplit = (FileSplit) inputSplit;
+ job.set(JobContext.MAP_INPUT_FILE, fileSplit.getPath().toString());
+ job.setLong(JobContext.MAP_INPUT_START, fileSplit.getStart());
+ job.setLong(JobContext.MAP_INPUT_PATH, fileSplit.getLength());
+ }
+ LOG.info("Processing split: " + inputSplit);
+ }
+
+ private long getInputBytes() {
+ if (fsStats == null) return 0;
+ long bytesRead = 0;
+ for (Statistics stat: fsStats) {
+ bytesRead = bytesRead + stat.getBytesRead();
+ }
+ return bytesRead;
+ }
+
+ public void initializeNewRecordReader(
+ org.apache.hadoop.mapreduce.InputSplit split, TaskAttemptContext context)
+ throws IOException, InterruptedException {
+ newRecordReader.initialize(split, context);
+ }
+
+ public org.apache.hadoop.mapreduce.InputSplit getNewInputSplit() {
+ return newInputSplit;
+ }
+
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/SimpleOutput.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/SimpleOutput.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/SimpleOutput.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/SimpleOutput.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,199 @@
+/**
+ * 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.output;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystem.Statistics;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.api.Master;
+import org.apache.tez.api.Output;
+import org.apache.tez.common.TezTask;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.processor.MRTask;
+import org.apache.tez.mapreduce.processor.MRTaskReporter;
+import org.apache.tez.records.OutputContext;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+
+/**
+ * {@link SimpleOutput} is an {@link Output} which persists key/value pairs
+ * written to it.
+ *
+ * It is compatible with all standard Apache Hadoop MapReduce
+ * {@link OutputFormat} implementations.
+ */
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class SimpleOutput implements Output {
+
+ private MRTask task;
+
+ boolean useNewApi;
+ JobConf jobConf;
+
+ org.apache.hadoop.mapreduce.TaskAttemptContext taskAttemptContext;
+ org.apache.hadoop.mapreduce.OutputFormat newOutputFormat;
+ org.apache.hadoop.mapreduce.RecordWriter newRecordWriter;
+
+ org.apache.hadoop.mapred.OutputFormat oldOutputFormat;
+ org.apache.hadoop.mapred.RecordWriter oldRecordWriter;
+
+ private TezCounter outputRecordCounter;
+ private TezCounter fileOutputByteCounter;
+ private List<Statistics> fsStats;
+ private MRTaskReporter reporter;
+
+ @Inject
+ public SimpleOutput(
+ @Assisted TezTask task
+ ) {
+ }
+
+ public void setTask(MRTask task) {
+ this.task = task;
+ }
+
+ public void initialize(Configuration conf, Master master) throws IOException,
+ InterruptedException {
+
+ if (task == null) {
+ return;
+ }
+
+ if (conf instanceof JobConf) {
+ jobConf = (JobConf)conf;
+ } else {
+ jobConf = new JobConf(conf);
+ }
+
+ useNewApi = jobConf.getUseNewMapper();
+ taskAttemptContext = task.getTaskAttemptContext();
+
+ outputRecordCounter = task.getOutputRecordsCounter();
+ fileOutputByteCounter = task.getFileOutputBytesCounter();
+
+ reporter = task.getMRReporter();
+
+ if (useNewApi) {
+ try {
+ newOutputFormat =
+ ReflectionUtils.newInstance(
+ taskAttemptContext.getOutputFormatClass(), jobConf);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ }
+
+ List<Statistics> matchedStats = null;
+ if (newOutputFormat instanceof
+ org.apache.hadoop.mapreduce.lib.output.FileOutputFormat) {
+ matchedStats =
+ MRTask.getFsStatistics(
+ org.apache.hadoop.mapreduce.lib.output.FileOutputFormat
+ .getOutputPath(taskAttemptContext),
+ jobConf);
+ }
+ fsStats = matchedStats;
+
+ long bytesOutPrev = getOutputBytes();
+ newRecordWriter =
+ newOutputFormat.getRecordWriter(this.taskAttemptContext);
+ long bytesOutCurr = getOutputBytes();
+ fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev);
+ } else {
+ oldOutputFormat = jobConf.getOutputFormat();
+
+ List<Statistics> matchedStats = null;
+ if (oldOutputFormat instanceof org.apache.hadoop.mapred.FileOutputFormat) {
+ matchedStats =
+ MRTask.getFsStatistics(
+ org.apache.hadoop.mapred.FileOutputFormat.getOutputPath(
+ jobConf),
+ jobConf);
+ }
+ fsStats = matchedStats;
+
+ FileSystem fs = FileSystem.get(jobConf);
+ String finalName = task.getOutputName();
+
+ long bytesOutPrev = getOutputBytes();
+ oldRecordWriter =
+ oldOutputFormat.getRecordWriter(
+ fs, jobConf, finalName, reporter);
+ long bytesOutCurr = getOutputBytes();
+ fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev);
+ }
+ }
+
+ public void write(Object key, Object value)
+ throws IOException, InterruptedException {
+
+ reporter.progress();
+ long bytesOutPrev = getOutputBytes();
+
+ if (useNewApi) {
+ newRecordWriter.write(key, value);
+ } else {
+ oldRecordWriter.write(key, value);
+ }
+
+ long bytesOutCurr = getOutputBytes();
+ fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev);
+ outputRecordCounter.increment(1);
+
+ }
+
+ public void close() throws IOException, InterruptedException {
+ reporter.progress();
+ long bytesOutPrev = getOutputBytes();
+ if (useNewApi) {
+ newRecordWriter.close(taskAttemptContext);
+ } else {
+ oldRecordWriter.close(null);
+ }
+ long bytesOutCurr = getOutputBytes();
+ fileOutputByteCounter.increment(bytesOutCurr - bytesOutPrev);
+ }
+
+ public org.apache.hadoop.mapreduce.OutputFormat getNewOutputFormat() {
+ return newOutputFormat;
+ }
+
+ public org.apache.hadoop.mapred.OutputFormat getOldOutputFormat() {
+ return oldOutputFormat;
+ }
+
+ private long getOutputBytes() {
+ if (fsStats == null) return 0;
+ long bytesWritten = 0;
+ for (Statistics stat: fsStats) {
+ bytesWritten = bytesWritten + stat.getBytesWritten();
+ }
+ return bytesWritten;
+ }
+
+ @Override
+ public OutputContext getOutputContext() {
+ return null;
+ }
+
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/partition/MRPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/partition/MRPartitioner.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/partition/MRPartitioner.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/partition/MRPartitioner.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,104 @@
+/**
+* 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.partition;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Partitioner;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.api.Master;
+import org.apache.tez.mapreduce.processor.MRTask;
+
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class MRPartitioner implements org.apache.tez.api.Partitioner {
+
+ static final Log LOG = LogFactory.getLog(MRPartitioner.class);
+ private final MRTask task;
+
+ JobConf jobConf;
+ boolean useNewApi;
+
+ org.apache.hadoop.mapred.Partitioner oldPartitioner;
+ org.apache.hadoop.mapreduce.Partitioner newPartitioner;
+
+ public MRPartitioner(MRTask task) {
+ this.task = task;
+ }
+
+ public void initialize(Configuration conf, Master master)
+ throws IOException, InterruptedException {
+ if (conf instanceof JobConf) {
+ jobConf = (JobConf)conf;
+ } else {
+ jobConf = new JobConf(conf);
+ }
+
+ useNewApi = jobConf.getUseNewMapper();
+ final int partitions = jobConf.getNumReduceTasks();
+ if (useNewApi) {
+ if (partitions > 1) {
+ try {
+ newPartitioner = (org.apache.hadoop.mapreduce.Partitioner)
+ ReflectionUtils.newInstance(
+ task.getJobContext().getPartitionerClass(), jobConf);
+ } catch (ClassNotFoundException cnfe) {
+ throw new IOException(cnfe);
+ }
+ } else {
+ newPartitioner = new org.apache.hadoop.mapreduce.Partitioner() {
+ @Override
+ public int getPartition(Object key, Object value, int numPartitions) {
+ return numPartitions - 1;
+ }
+ };
+ }
+ } else {
+ if (partitions > 1) {
+ oldPartitioner = (Partitioner)
+ ReflectionUtils.newInstance(jobConf.getPartitionerClass(), jobConf);
+ } else {
+ oldPartitioner = new Partitioner() {
+ @Override
+ public void configure(JobConf job) {}
+
+ @Override
+ public int getPartition(Object key, Object value, int numPartitions) {
+ return numPartitions - 1;
+ }
+ };
+ }
+
+ }
+
+ }
+
+ @Override
+ public int getPartition(Object key, Object value, int numPartitions) {
+ if (useNewApi) {
+ return newPartitioner.getPartition(key, value, numPartitions);
+ } else {
+ return oldPartitioner.getPartition(key, value, numPartitions);
+ }
+ }
+
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/FileSystemStatisticsUpdater.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/FileSystemStatisticsUpdater.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/FileSystemStatisticsUpdater.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/FileSystemStatisticsUpdater.java Fri Mar 15 21:26:36 2013
@@ -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.processor;
+
+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);
+ }
+ }
+
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/GcTimeUpdater.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/GcTimeUpdater.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/GcTimeUpdater.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/GcTimeUpdater.java Fri Mar 15 21:26:36 2013
@@ -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.processor;
+
+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());
+ }
+ }
+ }
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTask.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTask.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTask.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTask.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,706 @@
+/**
+* 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.processor;
+
+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.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
+import org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter;
+import org.apache.hadoop.mapreduce.lib.reduce.WrappedReducer;
+import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitIndex;
+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.api.Master;
+import org.apache.tez.common.Constants;
+import org.apache.tez.common.TezTask;
+import org.apache.tez.common.TezTaskStatus;
+import org.apache.tez.common.TezTaskStatus.Phase;
+import org.apache.tez.common.TezTaskStatus.State;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.engine.common.sort.impl.TezRawKeyValueIterator;
+import org.apache.tez.mapreduce.hadoop.*;
+import org.apache.tez.mapreduce.hadoop.mapred.TaskAttemptContextImpl;
+import org.apache.tez.mapreduce.hadoop.mapreduce.JobContextImpl;
+import org.apache.tez.mapreduce.hadoop.records.ProceedToCompletionResponse;
+import org.apache.tez.mapreduce.partition.MRPartitioner;
+import org.apache.tez.mapreduce.task.impl.MRTaskContext;
+import org.apache.tez.records.TezJobID;
+import org.apache.tez.records.TezTaskAttemptID;
+import org.apache.tez.records.OutputContext;
+
+public abstract class MRTask
+extends org.apache.tez.common.TezTask {
+
+ 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 = new TezCounters();
+ protected GcTimeUpdater gcUpdater;
+ private ResourceCalculatorProcessTree pTree;
+ private long initCpuCumulativeTime = 0;
+
+ /* 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);
+ }
+
+ private final static int MAX_RETRIES = 10;
+
+ /** The number of milliseconds between progress reports. */
+ public static final int PROGRESS_INTERVAL = 3000;
+
+ private MRTaskReporter mrReporter;
+
+ private final TaskSplitIndex splitMetaInfo;
+
+ /**
+ * A Map where Key-> URIScheme and value->FileSystemStatisticUpdater
+ */
+ private Map<String, FileSystemStatisticUpdater> statisticUpdaters =
+ new HashMap<String, FileSystemStatisticUpdater>();
+
+ public MRTask(TezTask context) {
+ super(context.getTaskAttemptId(), context.getUser(), context.getJobName(),
+ context.getTaskModuleClassName());
+ MRTaskContext tc = (MRTaskContext)context;
+ splitMetaInfo =
+ new TaskSplitIndex(tc.getSplitLocation(), tc.getSplitOffset());
+ // TODO TEZAM4 Figure out initialization / run sequence of Input, Process,
+ // Output. Phase is MR specific.
+ status =
+ new MRTaskStatus(
+ getTaskAttemptId(),
+ counters,
+ MRTaskType.fromString(getTaskAttemptId().getTaskType()) == MRTaskType.MAP ? Phase.MAP
+ : Phase.SHUFFLE
+ );
+ gcUpdater = new GcTimeUpdater(counters);
+ }
+
+ public void initialize(Configuration conf, Master master) throws IOException,
+ InterruptedException {
+ if (conf instanceof JobConf) {
+ this.jobConf = (JobConf)conf;
+ } else {
+ this.jobConf = new JobConf(conf);
+ }
+ reporter =
+ new TezTaskReporterImpl(this, (TezTaskUmbilicalProtocol)master);
+ ((TezTaskReporterImpl)reporter).startCommunicationThread();
+
+ jobConf.set(Constants.TEZ_ENGINE_TASK_ATTEMPT_ID,
+ getTaskAttemptId().toString());
+
+ initResourceCalculatorPlugin();
+
+ LOG.info("MRTask.inited: taskAttemptId = " + getTaskAttemptId().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 TezTaskUmbilicalProtocol getUmbilical() {
+ return ((TezTaskReporterImpl)reporter).getUmbilical();
+ }
+
+ public void initTask(JobConf job, TezJobID tezJobID,
+ MRTaskReporter mrReporter,
+ boolean useNewApi) throws IOException,
+ InterruptedException {
+ this.jobConf = job;
+ this.jobContext = new JobContextImpl(job, tezJobID, mrReporter);
+ this.taskAttemptContext =
+ new TaskAttemptContextImpl(job, getTaskAttemptId(), mrReporter);
+ this.mrReporter = mrReporter;
+
+ if (getState() == State.UNASSIGNED) {
+ setState(State.RUNNING);
+ }
+
+ // Save the committer
+ 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);
+
+ partitioner = new MRPartitioner(this);
+ ((MRPartitioner)partitioner).initialize(job, getTaskReporter());
+
+ localizeConfiguration(jobConf);
+ }
+
+ 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; }
+
+ /**
+ * Return current phase of the task.
+ * needs to be synchronized as communication thread sends the phase every second
+ * @return the curent phase of the task
+ */
+ public synchronized TezTaskStatus.Phase getPhase(){
+ return status.getPhase();
+ }
+
+ /**
+ * Set current phase of the task.
+ * @param phase task phase
+ */
+ protected synchronized void setPhase(TezTaskStatus.Phase phase){
+ status.setPhase(phase);
+ }
+
+ 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(getTaskAttemptId().getTaskID().getId());
+ }
+
+ public void waitBeforeCompletion(MRTaskReporter reporter) throws IOException,
+ InterruptedException {
+ TezTaskUmbilicalProtocol umbilical = getUmbilical();
+ int retries = MAX_RETRIES;
+ boolean readyToProceed = false;
+ while (!readyToProceed) {
+ try {
+ ProceedToCompletionResponse response =
+ umbilical.proceedToCompletion(getTaskAttemptId());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Got readyToProceed: " + response);
+ }
+ if (response.shouldDie()) {
+ throw new IOException("Task was asked to die by the AM");
+ // TODO EVENTUALLY Figure out a good way for a graceful exit, instead
+ // of an exit via an Exception. This isn' necessarily an error.
+ }
+ readyToProceed = response.readyToProceed();
+ } catch (IOException ie) {
+ LOG.warn("Failure waiting for exit signal: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ throw ie;
+ }
+ }
+ synchronized(this) {
+ wait(1000l); // Check if ready to exit every second.
+ }
+ }
+ }
+
+ public void outputReady(MRTaskReporter reporter, OutputContext outputContext)
+ throws IOException,
+ InterruptedException {
+ LOG.info("Task: " + getTaskAttemptId() + " reporting outputReady");
+ updateCounters();
+ statusUpdate();
+
+ TezTaskUmbilicalProtocol umbilical = getUmbilical();
+ int retries = MAX_RETRIES;
+ while (true) {
+ try {
+ umbilical.outputReady(getTaskAttemptId(), outputContext);
+ LOG.info("Task '" + getTaskAttemptId() + "' reported outputReady.");
+ return;
+ } catch (IOException ie) {
+ LOG.warn("Failure signalling outputReady: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ throw ie;
+ }
+ }
+ }
+ }
+
+ public void done(
+ OutputContext outputContext,
+ MRTaskReporter reporter
+ ) throws IOException, InterruptedException {
+ updateCounters();
+ if (outputContext != null) {
+ LOG.info("Task: "
+ + getTaskAttemptId()
+ + " is done."
+ + " And is in the process of sending output-context with shuffle port: "
+ + outputContext.getShufflePort());
+ outputReady(reporter, outputContext);
+ waitBeforeCompletion(reporter);
+ }
+
+ LOG.info("Task:" + getTaskAttemptId() + " is done."
+ + " And is in the process of committing");
+ TezTaskUmbilicalProtocol umbilical = getUmbilical();
+ // TODO TEZ Interaciton between Commit and OutputReady. Merge ?
+ if (isCommitRequired()) {
+ 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(getTaskAttemptId(), 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(umbilical, reporter, committer);
+ }
+ taskDone.set(true);
+ reporter.stopCommunicationThread();
+ // 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 {
+ int retries = MAX_RETRIES;
+ while (true) {
+ try {
+ if (!getUmbilical().statusUpdate(getTaskAttemptId(), status)) {
+ LOG.warn("Parent died. Exiting " + getTaskAttemptId());
+ System.exit(66);
+ }
+ status.clearStatus();
+ return;
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt(); // interrupt ourself
+ } catch (IOException ie) {
+ LOG.warn("Failure sending status update: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ throw ie;
+ }
+ }
+ }
+ }
+
+ /**
+ * Sends last status update before sending umbilical.done();
+ */
+ private void sendLastUpdate()
+ throws IOException, InterruptedException {
+ status.setOutputSize(-1l);
+ // send a final status report
+ status.statusUpdate(
+ getProgress().get(), getProgress().toString(), counters);
+ statusUpdate();
+ }
+
+ private void commit(TezTaskUmbilicalProtocol umbilical,
+ MRTaskReporter reporter,
+ org.apache.hadoop.mapreduce.OutputCommitter committer
+ ) throws IOException {
+ int retries = MAX_RETRIES;
+ while (true) {
+ try {
+ while (!umbilical.canCommit(getTaskAttemptId())) {
+ // 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
+ }
+ reporter.setProgressFlag();
+ }
+ break;
+ } catch (IOException ie) {
+ LOG.warn("Failure asking whether task can commit: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ //if it couldn't query successfully then delete the output
+ discardOutput(taskAttemptContext);
+ System.exit(68);
+ }
+ }
+ }
+
+ // task can Commit now
+ try {
+ LOG.info("Task " + getTaskAttemptId() + " 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));
+ }
+ }
+
+
+ private void sendDone(TezTaskUmbilicalProtocol umbilical) throws IOException {
+ int retries = MAX_RETRIES;
+ while (true) {
+ try {
+ umbilical.done(getTaskAttemptId());
+ LOG.info("Task '" + getTaskAttemptId() + "' done.");
+ return;
+ } catch (IOException ie) {
+ LOG.warn("Failure signalling completion: " +
+ StringUtils.stringifyException(ie));
+ if (--retries == 0) {
+ throw ie;
+ }
+ }
+ }
+ }
+
+ 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,
+ TezTaskAttemptID 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,
+ IDConverter.toMRTaskAttemptId(taskId),
+ r,
+ inputKeyCounter,
+ inputValueCounter,
+ output,
+ committer,
+ reporter,
+ comparator,
+ keyClass,
+ 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(TezTaskUmbilicalProtocol umbilical)
+ throws IOException, InterruptedException {
+ // set phase for this task
+ setPhase(TezTaskStatus.Phase.CLEANUP);
+ getProgress().setStatus("cleanup");
+ 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, getTaskAttemptId().getTaskID().toString());
+ jobConf.set(JobContext.TASK_ATTEMPT_ID, getTaskAttemptId().toString());
+ jobConf.setInt(JobContext.TASK_PARTITION,
+ getTaskAttemptId().getTaskID().getId());
+ jobConf.set(JobContext.ID, getTaskAttemptId().getJobID().toString());
+ }
+
+ public abstract TezCounter getOutputRecordsCounter();
+
+ public abstract TezCounter getInputRecordsCounter();
+
+ public TezCounter getFileOutputBytesCounter() {
+ return reporter.getCounter(FileOutputFormatCounter.BYTES_WRITTEN);
+ }
+
+ public org.apache.hadoop.mapreduce.TaskAttemptContext getTaskAttemptContext() {
+ return taskAttemptContext;
+ }
+
+ public TezCounter getFileInputBytesCounter() {
+ return reporter.getCounter(FileInputFormatCounter.BYTES_READ);
+ }
+
+ public TaskSplitIndex getSplitIndex() {
+ return splitMetaInfo;
+ }
+
+ public JobContext getJobContext() {
+ return jobContext;
+ }
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTaskReporter.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTaskReporter.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTaskReporter.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/MRTaskReporter.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,111 @@
+/**
+* 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.processor;
+
+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.TezTaskReporter;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.hadoop.mapred.MRCounters;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class MRTaskReporter
+ extends org.apache.hadoop.mapreduce.StatusReporter
+ implements Reporter {
+
+ private final TezTaskReporterImpl reporter;
+
+ private InputSplit split = null;
+
+ public MRTaskReporter(TezTaskReporter reporter) {
+ this.reporter = (TezTaskReporterImpl)reporter;
+ }
+
+ // getters and setters for flag
+ void setProgressFlag() {
+ reporter.setProgressFlag();
+ }
+ boolean resetProgressFlag() {
+ return reporter.resetProgressFlag();
+ }
+ public void setStatus(String status) {
+ reporter.setStatus(status);
+ }
+ public void setProgress(float progress) {
+ reporter.setProgress(progress);
+ }
+
+ public float getProgress() {
+ return reporter.getProgress();
+ };
+
+ public void progress() {
+ reporter.progress();
+ }
+
+ public Counters.Counter getCounter(String group, String name) {
+ TezCounter counter = reporter.getCounter(group, name);
+ MRCounters.MRCounter mrCounter = null;
+ if (counter != null) {
+ mrCounter = new MRCounters.MRCounter(counter);
+ }
+ return mrCounter;
+ }
+
+ public Counters.Counter getCounter(Enum<?> name) {
+ TezCounter counter = reporter.getCounter(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;
+ }
+ }
+
+ public void startCommunicationThread() {
+ reporter.startCommunicationThread();
+ }
+
+ public void stopCommunicationThread() throws InterruptedException {
+ reporter.stopCommunicationThread();
+ }
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/TezTaskReporterImpl.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/TezTaskReporterImpl.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/TezTaskReporterImpl.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/TezTaskReporterImpl.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,271 @@
+/**
+* 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.processor;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.ipc.ProtocolSignature;
+import org.apache.hadoop.util.Progress;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.hadoop.TezTaskUmbilicalProtocol;
+import org.apache.tez.records.TezJobID;
+import org.apache.tez.records.TezTaskAttemptID;
+import org.apache.tez.records.TezTaskDependencyCompletionEventsUpdate;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+class TezTaskReporterImpl
+ implements org.apache.tez.common.TezTaskReporter, Runnable {
+
+ private static final Log LOG = LogFactory.getLog(TezTaskReporterImpl.class);
+
+ private final MRTask mrTask;
+ private final TezTaskUmbilicalProtocol umbilical;
+ private final Progress taskProgress;
+
+ private Thread pingThread = null;
+ private boolean done = true;
+ private Object lock = new Object();
+
+ /**
+ * flag that indicates whether progress update needs to be sent to parent.
+ * If true, it has been set. If false, it has been reset.
+ * Using AtomicBoolean since we need an atomic read & reset method.
+ */
+ private AtomicBoolean progressFlag = new AtomicBoolean(false);
+
+ TezTaskReporterImpl(MRTask mrTask, TezTaskUmbilicalProtocol umbilical) {
+ this.mrTask = mrTask;
+ this.umbilical = umbilical;
+ this.taskProgress = mrTask.getProgress();
+ }
+
+ // getters and setters for flag
+ void setProgressFlag() {
+ progressFlag.set(true);
+ }
+
+ boolean resetProgressFlag() {
+ return progressFlag.getAndSet(false);
+ }
+
+ public void setStatus(String status) {
+ // FIXME - BADLY
+ if (true) {
+ return;
+ }
+ taskProgress.setStatus(
+ MRTask.normalizeStatus(status, this.mrTask.jobConf));
+ // indicate that progress update needs to be sent
+ setProgressFlag();
+ }
+
+ public void setProgress(float progress) {
+ // set current phase progress.
+ // This method assumes that task has phases.
+ taskProgress.phase().set(progress);
+ // indicate that progress update needs to be sent
+ setProgressFlag();
+ }
+
+ public float getProgress() {
+ return taskProgress.getProgress();
+ };
+
+ public void progress() {
+ // indicate that progress update needs to be sent
+ setProgressFlag();
+ }
+
+ public TezCounter getCounter(String group, String name) {
+ return this.mrTask.counters == null ?
+ null :
+ this.mrTask.counters.findCounter(group, name);
+ }
+
+ public TezCounter getCounter(Enum<?> name) {
+ return this.mrTask.counters == null ?
+ null :
+ this.mrTask.counters.findCounter(name);
+ }
+
+ public void incrCounter(Enum<?> key, long amount) {
+ if (this.mrTask.counters != null) {
+ this.mrTask.counters.findCounter(key).increment(amount);
+ }
+ setProgressFlag();
+ }
+
+ public void incrCounter(String group, String counter, long amount) {
+ if (this.mrTask.counters != null) {
+ this.mrTask.counters.findCounter(group, counter).increment(amount);
+ }
+ setProgressFlag();
+ }
+
+ /**
+ * The communication thread handles communication with the parent (Task Tracker).
+ * It sends progress updates if progress has been made or if the task needs to
+ * let the parent know that it's alive. It also pings the parent to see if it's alive.
+ */
+ public void run() {
+ final int MAX_RETRIES = 3;
+ int remainingRetries = MAX_RETRIES;
+ // get current flag value and reset it as well
+ boolean sendProgress = resetProgressFlag();
+ while (!this.mrTask.taskDone.get()) {
+ synchronized (lock) {
+ done = false;
+ }
+ try {
+ boolean taskFound = true; // whether TT knows about this task
+ // sleep for a bit
+ synchronized(lock) {
+ if (this.mrTask.taskDone.get()) {
+ break;
+ }
+ lock.wait(MRTask.PROGRESS_INTERVAL);
+ }
+ if (this.mrTask.taskDone.get()) {
+ break;
+ }
+
+ if (sendProgress) {
+ // we need to send progress update
+ this.mrTask.updateCounters();
+ this.mrTask.getStatus().statusUpdate(
+ taskProgress.get(),
+ taskProgress.toString(),
+ this.mrTask.counters);
+ taskFound =
+ umbilical.statusUpdate(
+ this.mrTask.getTaskAttemptId(), this.mrTask.getStatus());
+ this.mrTask.getStatus().clearStatus();
+ }
+ else {
+ // send ping
+ taskFound = umbilical.ping(this.mrTask.getTaskAttemptId());
+ }
+
+ // if Task Tracker is not aware of our task ID (probably because it died and
+ // came back up), kill ourselves
+ if (!taskFound) {
+ MRTask.LOG.warn("Parent died. Exiting " + this.mrTask.getTaskAttemptId());
+ resetDoneFlag();
+ System.exit(66);
+ }
+
+ sendProgress = resetProgressFlag();
+ remainingRetries = MAX_RETRIES;
+ }
+ catch (Throwable t) {
+ MRTask.LOG.info("Communication exception: " + StringUtils.stringifyException(t));
+ remainingRetries -=1;
+ if (remainingRetries == 0) {
+ ReflectionUtils.logThreadInfo(MRTask.LOG, "Communication exception", 0);
+ MRTask.LOG.warn("Last retry, killing " + this.mrTask.getTaskAttemptId());
+ resetDoneFlag();
+ System.exit(65);
+ }
+ }
+ }
+ //Notify that we are done with the work
+ resetDoneFlag();
+ }
+ void resetDoneFlag() {
+ synchronized (lock) {
+ done = true;
+ lock.notify();
+ }
+ }
+ public void startCommunicationThread() {
+ if (pingThread == null) {
+ pingThread = new Thread(this, "communication thread");
+ pingThread.setDaemon(true);
+ pingThread.start();
+ }
+ }
+ public void stopCommunicationThread() throws InterruptedException {
+ if (pingThread != null) {
+ // Intent of the lock is to not send an interupt in the middle of an
+ // umbilical.ping or umbilical.statusUpdate
+ synchronized(lock) {
+ //Interrupt if sleeping. Otherwise wait for the RPC call to return.
+ lock.notify();
+ }
+
+ synchronized (lock) {
+ while (!done) {
+ lock.wait();
+ }
+ }
+ pingThread.interrupt();
+ pingThread.join();
+ }
+ }
+
+ @Override
+ public TezTaskDependencyCompletionEventsUpdate getDependentTasksCompletionEvents(
+ TezJobID jobID, int fromEventIdx, int maxEventsToFetch,
+ TezTaskAttemptID reduce) {
+ return umbilical.getDependentTasksCompletionEvents(
+ jobID, fromEventIdx, maxEventsToFetch, reduce);
+ }
+
+ @Override
+ public void reportFatalError(TezTaskAttemptID taskAttemptId,
+ Throwable throwable, String logMsg) {
+ LOG.fatal(logMsg);
+ Throwable tCause = throwable.getCause();
+ String cause = tCause == null
+ ? StringUtils.stringifyException(throwable)
+ : StringUtils.stringifyException(tCause);
+ try {
+ umbilical.fatalError(mrTask.getTaskAttemptId(), cause);
+ } catch (IOException ioe) {
+ LOG.fatal("Failed to contact the tasktracker", ioe);
+ System.exit(-1);
+ }
+ }
+
+ public TezTaskUmbilicalProtocol getUmbilical() {
+ return umbilical;
+ }
+
+ @Override
+ public long getProtocolVersion(String protocol, long clientVersion)
+ throws IOException {
+ // TODO TEZAM3
+ return 1;
+ }
+
+ @Override
+ public ProtocolSignature getProtocolSignature(String protocol,
+ long clientVersion, int clientMethodsHash) throws IOException {
+ return ProtocolSignature.getProtocolSignature(this, protocol,
+ clientVersion, clientMethodsHash);
+ }
+}
Added: incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/map/MapProcessor.java
URL: http://svn.apache.org/viewvc/incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/map/MapProcessor.java?rev=1457129&view=auto
==============================================================================
--- incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/map/MapProcessor.java (added)
+++ incubator/tez/tez-mapreduce/src/main/java/org/apache/tez/mapreduce/processor/map/MapProcessor.java Fri Mar 15 21:26:36 2013
@@ -0,0 +1,362 @@
+/**
+ * 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.processor.map;
+
+import java.io.IOException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+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.mapreduce.split.JobSplit.TaskSplitIndex;
+import org.apache.hadoop.util.Progress;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.api.Input;
+import org.apache.tez.api.Master;
+import org.apache.tez.api.Output;
+import org.apache.tez.api.Processor;
+import org.apache.tez.common.TezTask;
+import org.apache.tez.common.TezTaskStatus;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.common.counters.TaskCounter;
+import org.apache.tez.engine.common.sort.SortingOutput;
+import org.apache.tez.mapreduce.hadoop.IDConverter;
+import org.apache.tez.mapreduce.hadoop.mapred.TaskAttemptContextImpl;
+import org.apache.tez.mapreduce.input.SimpleInput;
+import org.apache.tez.mapreduce.output.SimpleOutput;
+import org.apache.tez.mapreduce.processor.MRTask;
+import org.apache.tez.mapreduce.processor.MRTaskReporter;
+import org.apache.tez.records.OutputContext;
+
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+
+@SuppressWarnings({ "unchecked", "rawtypes" })
+public class MapProcessor extends MRTask implements Processor {
+
+ private static final Log LOG = LogFactory.getLog(MapProcessor.class);
+
+ private Progress mapPhase;
+
+ @Inject
+ public MapProcessor(
+ @Assisted TezTask context
+ ) {
+ super(context);
+ }
+
+ @Override
+ public void initialize(Configuration conf, Master master) throws IOException,
+ InterruptedException {
+ super.initialize(conf, master);
+ }
+
+ @Override
+ public void process(
+ final Input in,
+ final Output out)
+ throws IOException, InterruptedException {
+ MRTaskReporter reporter = new MRTaskReporter(getTaskReporter());
+ boolean useNewApi = jobConf.getUseNewMapper();
+ initTask(jobConf, getJobID(), reporter, useNewApi);
+
+ if (in instanceof SimpleInput) {
+ ((SimpleInput)in).setTask(this);
+ }
+
+ if (out instanceof SimpleOutput) {
+ ((SimpleOutput)out).setTask(this);
+ } else if (out instanceof SortingOutput) {
+ ((SortingOutput)out).setTask(this);
+ }
+
+
+ in.initialize(jobConf, getTaskReporter());
+ out.initialize(jobConf, getTaskReporter());
+
+ // If there are no reducers then there won't be any sort. Hence the map
+ // phase will govern the entire attempt's progress.
+ if (jobConf.getNumReduceTasks() == 0) {
+ mapPhase = getProgress().addPhase("map", 1.0f);
+ } else {
+ // If there are reducers then the entire attempt's progress will be
+ // split between the map phase (67%) and the sort phase (33%).
+ mapPhase = getProgress().addPhase("map", 0.667f);
+ }
+
+ // Sanity check
+ if (!(in instanceof SimpleInput)) {
+ throw new IOException("Unknown input! - " + in.getClass());
+ }
+ SimpleInput input = (SimpleInput)in;
+
+ if (useNewApi) {
+ runNewMapper(jobConf, getSplitIndex(), reporter, input, out, getTaskReporter());
+ } else {
+ runOldMapper(jobConf, getSplitIndex(), reporter, input, out, getTaskReporter());
+ }
+
+ done(out.getOutputContext(), reporter);
+ }
+
+ public void close() throws IOException, InterruptedException {
+ // TODO Auto-generated method stub
+
+ }
+
+ void runOldMapper(
+ final JobConf job,
+ final TaskSplitIndex splitIndex,
+ final MRTaskReporter reporter,
+ final SimpleInput input,
+ final Output output,
+ final Master master
+ ) throws IOException, InterruptedException {
+
+ RecordReader in = new OldRecordReader(input);
+
+ int numReduceTasks = job.getNumReduceTasks();
+ LOG.info("numReduceTasks: " + numReduceTasks);
+
+ OutputCollector collector = new OldOutputCollector(output);
+
+ MapRunnable runner =
+ (MapRunnable)ReflectionUtils.newInstance(job.getMapRunnerClass(), job);
+
+ try {
+ runner.run(in, collector, (Reporter)reporter);
+ mapPhase.complete();
+ // start the sort phase only if there are reducers
+ if (numReduceTasks > 0) {
+ setPhase(TezTaskStatus.Phase.SORT);
+ }
+ this.statusUpdate();
+ } finally {
+ //close
+ in.close(); // close input
+ output.close();
+ }
+ }
+
+ private void runNewMapper(final JobConf job,
+ final TaskSplitIndex splitIndex,
+ MRTaskReporter reporter,
+ final SimpleInput in,
+ Output out,
+ final Master master
+ ) throws IOException, InterruptedException {
+ // make a task context so we can get the classes
+ org.apache.hadoop.mapreduce.TaskAttemptContext taskContext =
+ new TaskAttemptContextImpl(job, getTaskAttemptId(), reporter);
+
+ // 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);
+ }
+
+ if (!(in instanceof SimpleInput)) {
+ throw new IOException("Unknown input! - " + in.getClass());
+ }
+
+ 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 org.apache.tez.mapreduce.hadoop.mapreduce.MapContextImpl(
+ job, IDConverter.toMRTaskAttemptId(getTaskAttemptId()),
+ input, output,
+ getCommitter(),
+ reporter, split);
+
+ org.apache.hadoop.mapreduce.Mapper.Context mapperContext =
+ new WrappedMapper().getMapContext(mapContext);
+
+ input.initialize(split, mapperContext);
+ mapper.run(mapperContext);
+ mapPhase.complete();
+ setPhase(TezTaskStatus.Phase.SORT);
+ this.statusUpdate();
+ input.close();
+ output.close(mapperContext);
+ }
+
+ private static class NewRecordReader extends
+ org.apache.hadoop.mapreduce.RecordReader {
+ private final SimpleInput in;
+
+ private NewRecordReader(SimpleInput in) {
+ this.in = in;
+ }
+
+ @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 in.hasNext();
+ }
+
+ @Override
+ public Object getCurrentKey() throws IOException,
+ InterruptedException {
+ return in.getNextKey();
+ }
+
+ @Override
+ public Object getCurrentValue() throws IOException,
+ InterruptedException {
+ return in.getNextValues().iterator().next();
+ }
+
+ @Override
+ public float getProgress() throws IOException, InterruptedException {
+ return in.getProgress();
+ }
+
+ @Override
+ public void close() throws IOException {
+ in.close();
+ }
+ }
+
+ private static class OldRecordReader implements RecordReader {
+ private final SimpleInput simpleInput;
+
+ private OldRecordReader(SimpleInput simpleInput) {
+ this.simpleInput = simpleInput;
+ }
+
+ @Override
+ public boolean next(Object key, Object value) throws IOException {
+ simpleInput.setKey(key);
+ simpleInput.setValue(value);
+ try {
+ return simpleInput.hasNext();
+ } catch (InterruptedException ie) {
+ throw new IOException(ie);
+ }
+ }
+
+ @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 {
+ simpleInput.close();
+ }
+
+ @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 Output output;
+
+ OldOutputCollector(Output output) {
+ this.output = output;
+ }
+
+ public void collect(Object key, Object value) throws IOException {
+ try {
+ output.write(key, value);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ throw new IOException("interrupt exception", ie);
+ }
+ }
+ }
+
+ private class NewOutputCollector
+ extends org.apache.hadoop.mapreduce.RecordWriter {
+ private final Output out;
+
+ NewOutputCollector(Output 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 {
+ out.close();
+ }
+ }
+
+ @Override
+ public void localizeConfiguration(JobConf jobConf)
+ throws IOException, InterruptedException {
+ super.localizeConfiguration(jobConf);
+ jobConf.setBoolean(JobContext.TASK_ISMAP, true);
+ }
+
+ @Override
+ public TezCounter getOutputRecordsCounter() {
+ return reporter.getCounter(TaskCounter.MAP_OUTPUT_RECORDS);
+ }
+
+ @Override
+ public TezCounter getInputRecordsCounter() {
+ return reporter.getCounter(TaskCounter.MAP_INPUT_RECORDS);
+
+ }
+}