MAPREDUCE-3697. Support binary compatibility for Counters after MAPREDUCE-901.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1241319 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
2d1406e9e7
commit
35f12b9556
|
@ -736,6 +736,9 @@ Release 0.23.1 - Unreleased
|
||||||
MAPREDUCE-3794. Support mapred.Task.Counter and mapred.JobInProgress.Counter
|
MAPREDUCE-3794. Support mapred.Task.Counter and mapred.JobInProgress.Counter
|
||||||
enums for compatibility (Tom White via mahadev)
|
enums for compatibility (Tom White via mahadev)
|
||||||
|
|
||||||
|
MAPREDUCE-3697. Support binary compatibility for Counters after
|
||||||
|
MAPREDUCE-901. (mahadev via acmurthy)
|
||||||
|
|
||||||
Release 0.23.0 - 2011-11-01
|
Release 0.23.0 - 2011-11-01
|
||||||
|
|
||||||
INCOMPATIBLE CHANGES
|
INCOMPATIBLE CHANGES
|
||||||
|
|
|
@ -21,8 +21,14 @@ package org.apache.hadoop.mapred;
|
||||||
import static org.apache.hadoop.mapreduce.util.CountersStrings.parseEscapedCompactString;
|
import static org.apache.hadoop.mapreduce.util.CountersStrings.parseEscapedCompactString;
|
||||||
import static org.apache.hadoop.mapreduce.util.CountersStrings.toEscapedCompactString;
|
import static org.apache.hadoop.mapreduce.util.CountersStrings.toEscapedCompactString;
|
||||||
|
|
||||||
|
import java.io.DataInput;
|
||||||
|
import java.io.DataOutput;
|
||||||
|
import java.io.IOException;
|
||||||
import java.text.ParseException;
|
import java.text.ParseException;
|
||||||
|
import java.util.Collection;
|
||||||
|
import java.util.Iterator;
|
||||||
|
|
||||||
|
import org.apache.commons.collections.IteratorUtils;
|
||||||
import org.apache.commons.logging.Log;
|
import org.apache.commons.logging.Log;
|
||||||
import org.apache.hadoop.classification.InterfaceAudience;
|
import org.apache.hadoop.classification.InterfaceAudience;
|
||||||
import org.apache.hadoop.classification.InterfaceStability;
|
import org.apache.hadoop.classification.InterfaceStability;
|
||||||
|
@ -36,6 +42,9 @@ import org.apache.hadoop.mapreduce.counters.FrameworkCounterGroup;
|
||||||
import org.apache.hadoop.mapreduce.counters.GenericCounter;
|
import org.apache.hadoop.mapreduce.counters.GenericCounter;
|
||||||
import org.apache.hadoop.mapreduce.counters.Limits;
|
import org.apache.hadoop.mapreduce.counters.Limits;
|
||||||
import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
|
import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
|
||||||
|
import org.apache.hadoop.mapreduce.util.CountersStrings;
|
||||||
|
|
||||||
|
import com.google.common.collect.Iterators;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A set of named counters.
|
* A set of named counters.
|
||||||
|
@ -52,6 +61,8 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
|
||||||
public class Counters
|
public class Counters
|
||||||
extends AbstractCounters<Counters.Counter, Counters.Group> {
|
extends AbstractCounters<Counters.Counter, Counters.Group> {
|
||||||
|
|
||||||
|
public static int MAX_COUNTER_LIMIT = Limits.COUNTERS_MAX;
|
||||||
|
|
||||||
public Counters() {
|
public Counters() {
|
||||||
super(groupFactory);
|
super(groupFactory);
|
||||||
}
|
}
|
||||||
|
@ -69,17 +80,82 @@ public class Counters
|
||||||
return new Counters(newCounters);
|
return new Counters(newCounters);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public synchronized Group getGroup(String groupName) {
|
||||||
|
return super.getGroup(groupName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
public synchronized Collection<String> getGroupNames() {
|
||||||
|
return IteratorUtils.toList(super.getGroupNames().iterator());
|
||||||
|
}
|
||||||
|
|
||||||
|
public synchronized String makeCompactString() {
|
||||||
|
return CountersStrings.toEscapedCompactString(this);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A counter record, comprising its name and value.
|
* A counter record, comprising its name and value.
|
||||||
*/
|
*/
|
||||||
public interface Counter extends org.apache.hadoop.mapreduce.Counter {
|
public static class Counter implements org.apache.hadoop.mapreduce.Counter {
|
||||||
|
org.apache.hadoop.mapreduce.Counter realCounter;
|
||||||
|
|
||||||
|
Counter(org.apache.hadoop.mapreduce.Counter counter) {
|
||||||
|
this.realCounter = counter;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Counter() {
|
||||||
|
this(new GenericCounter());
|
||||||
|
}
|
||||||
|
|
||||||
|
@SuppressWarnings("deprecation")
|
||||||
|
@Override
|
||||||
|
public void setDisplayName(String displayName) {
|
||||||
|
realCounter.setDisplayName(displayName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getName() {
|
||||||
|
return realCounter.getName();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getDisplayName() {
|
||||||
|
return realCounter.getDisplayName();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public long getValue() {
|
||||||
|
return realCounter.getValue();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setValue(long value) {
|
||||||
|
realCounter.setValue(value);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void increment(long incr) {
|
||||||
|
realCounter.increment(incr);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void write(DataOutput out) throws IOException {
|
||||||
|
realCounter.write(out);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void readFields(DataInput in) throws IOException {
|
||||||
|
realCounter.readFields(in);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Returns the compact stringified version of the counter in the format
|
* Returns the compact stringified version of the counter in the format
|
||||||
* [(actual-name)(display-name)(value)]
|
* [(actual-name)(display-name)(value)]
|
||||||
* @return the stringified result
|
* @return the stringified result
|
||||||
*/
|
*/
|
||||||
String makeEscapedCompactString();
|
public String makeEscapedCompactString() {
|
||||||
|
return toEscapedCompactString(realCounter);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Checks for (content) equality of two (basic) counters
|
* Checks for (content) equality of two (basic) counters
|
||||||
|
@ -88,38 +164,41 @@ public class Counters
|
||||||
* @deprecated
|
* @deprecated
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
@Deprecated
|
||||||
boolean contentEquals(Counter counter);
|
public boolean contentEquals(Counter counter) {
|
||||||
|
return realCounter.equals(counter.getUnderlyingCounter());
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @return the value of the counter
|
* @return the value of the counter
|
||||||
*/
|
*/
|
||||||
long getCounter();
|
|
||||||
}
|
|
||||||
|
|
||||||
static class OldCounterImpl extends GenericCounter implements Counter {
|
|
||||||
|
|
||||||
OldCounterImpl() {
|
|
||||||
}
|
|
||||||
|
|
||||||
OldCounterImpl(String name, String displayName, long value) {
|
|
||||||
super(name, displayName, value);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public synchronized String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override @Deprecated
|
|
||||||
public boolean contentEquals(Counter counter) {
|
|
||||||
return equals(counter);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public long getCounter() {
|
public long getCounter() {
|
||||||
return getValue();
|
return realCounter.getValue();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public org.apache.hadoop.mapreduce.Counter getUnderlyingCounter() {
|
||||||
|
return realCounter;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public synchronized boolean equals(Object genericRight) {
|
||||||
|
if (genericRight instanceof Counter) {
|
||||||
|
synchronized (genericRight) {
|
||||||
|
Counter right = (Counter) genericRight;
|
||||||
|
return getName().equals(right.getName()) &&
|
||||||
|
getDisplayName().equals(right.getDisplayName()) &&
|
||||||
|
getValue() == right.getValue();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public int hashCode() {
|
||||||
|
return realCounter.hashCode();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* <code>Group</code> of counters, comprising of counters from a particular
|
* <code>Group</code> of counters, comprising of counters from a particular
|
||||||
|
@ -128,21 +207,38 @@ public class Counters
|
||||||
* <p><code>Group</code>handles localization of the class name and the
|
* <p><code>Group</code>handles localization of the class name and the
|
||||||
* counter names.</p>
|
* counter names.</p>
|
||||||
*/
|
*/
|
||||||
public static interface Group extends CounterGroupBase<Counter> {
|
public static class Group implements CounterGroupBase<Counter> {
|
||||||
|
private CounterGroupBase<Counter> realGroup;
|
||||||
|
|
||||||
|
Group(GenericGroup group) {
|
||||||
|
this.realGroup = group;
|
||||||
|
}
|
||||||
|
Group(FSGroupImpl group) {
|
||||||
|
this.realGroup = group;
|
||||||
|
}
|
||||||
|
|
||||||
|
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||||
|
Group(FrameworkGroupImpl group) {
|
||||||
|
this.realGroup = group;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param counterName the name of the counter
|
* @param counterName the name of the counter
|
||||||
* @return the value of the specified counter, or 0 if the counter does
|
* @return the value of the specified counter, or 0 if the counter does
|
||||||
* not exist.
|
* not exist.
|
||||||
*/
|
*/
|
||||||
long getCounter(String counterName);
|
public long getCounter(String counterName) {
|
||||||
|
return getCounterValue(realGroup, counterName);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @return the compact stringified version of the group in the format
|
* @return the compact stringified version of the group in the format
|
||||||
* {(actual-name)(display-name)(value)[][][]} where [] are compact strings
|
* {(actual-name)(display-name)(value)[][][]} where [] are compact strings
|
||||||
* for the counters within.
|
* for the counters within.
|
||||||
*/
|
*/
|
||||||
String makeEscapedCompactString();
|
public String makeEscapedCompactString() {
|
||||||
|
return toEscapedCompactString(realGroup);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Get the counter for the given id and create it if it doesn't exist.
|
* Get the counter for the given id and create it if it doesn't exist.
|
||||||
|
@ -152,172 +248,184 @@ public class Counters
|
||||||
* @deprecated use {@link #findCounter(String)} instead
|
* @deprecated use {@link #findCounter(String)} instead
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
@Deprecated
|
||||||
Counter getCounter(int id, String name);
|
public Counter getCounter(int id, String name) {
|
||||||
|
return findCounter(name);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Get the counter for the given name and create it if it doesn't exist.
|
* Get the counter for the given name and create it if it doesn't exist.
|
||||||
* @param name the internal counter name
|
* @param name the internal counter name
|
||||||
* @return the counter
|
* @return the counter
|
||||||
*/
|
*/
|
||||||
Counter getCounterForName(String name);
|
public Counter getCounterForName(String name) {
|
||||||
|
return findCounter(name);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void write(DataOutput out) throws IOException {
|
||||||
|
realGroup.write(out);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void readFields(DataInput in) throws IOException {
|
||||||
|
realGroup.readFields(in);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Iterator<Counter> iterator() {
|
||||||
|
return realGroup.iterator();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getName() {
|
||||||
|
return realGroup.getName();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public String getDisplayName() {
|
||||||
|
return realGroup.getDisplayName();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setDisplayName(String displayName) {
|
||||||
|
realGroup.setDisplayName(displayName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void addCounter(Counter counter) {
|
||||||
|
realGroup.addCounter(counter);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter addCounter(String name, String displayName, long value) {
|
||||||
|
return realGroup.addCounter(name, displayName, value);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter findCounter(String counterName, String displayName) {
|
||||||
|
return realGroup.findCounter(counterName, displayName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter findCounter(String counterName, boolean create) {
|
||||||
|
return realGroup.findCounter(counterName, create);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter findCounter(String counterName) {
|
||||||
|
return realGroup.findCounter(counterName);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public int size() {
|
||||||
|
return realGroup.size();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void incrAllCounters(CounterGroupBase<Counter> rightGroup) {
|
||||||
|
realGroup.incrAllCounters(rightGroup);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return realGroup;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public synchronized boolean equals(Object genericRight) {
|
||||||
|
if (genericRight instanceof CounterGroupBase<?>) {
|
||||||
|
@SuppressWarnings("unchecked")
|
||||||
|
CounterGroupBase<Counter> right = ((CounterGroupBase<Counter>)
|
||||||
|
genericRight).getUnderlyingGroup();
|
||||||
|
return Iterators.elementsEqual(iterator(), right.iterator());
|
||||||
|
}
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public int hashCode() {
|
||||||
|
return realGroup.hashCode();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// All the group impls need this for legacy group interface
|
// All the group impls need this for legacy group interface
|
||||||
static long getCounterValue(Group group, String counterName) {
|
static long getCounterValue(CounterGroupBase<Counter> group, String counterName) {
|
||||||
Counter counter = group.findCounter(counterName, false);
|
Counter counter = group.findCounter(counterName, false);
|
||||||
if (counter != null) return counter.getValue();
|
if (counter != null) return counter.getValue();
|
||||||
return 0L;
|
return 0L;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Mix the generic group implementation into the Group interface
|
// Mix the generic group implementation into the Group interface
|
||||||
private static class GenericGroup extends AbstractCounterGroup<Counter>
|
private static class GenericGroup extends AbstractCounterGroup<Counter> {
|
||||||
implements Group {
|
|
||||||
|
|
||||||
GenericGroup(String name, String displayName, Limits limits) {
|
GenericGroup(String name, String displayName, Limits limits) {
|
||||||
super(name, displayName, limits);
|
super(name, displayName, limits);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public long getCounter(String counterName) {
|
|
||||||
return getCounterValue(this, counterName);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public Counter getCounter(int id, String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public Counter getCounterForName(String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Counter newCounter(String counterName, String displayName,
|
protected Counter newCounter(String counterName, String displayName,
|
||||||
long value) {
|
long value) {
|
||||||
return new OldCounterImpl(counterName, displayName, value);
|
return new Counter(new GenericCounter(counterName, displayName, value));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Counter newCounter() {
|
protected Counter newCounter() {
|
||||||
return new OldCounterImpl();
|
return new Counter();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return this;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Mix the framework group implementation into the Group interface
|
// Mix the framework group implementation into the Group interface
|
||||||
private static class FrameworkGroupImpl<T extends Enum<T>>
|
private static class FrameworkGroupImpl<T extends Enum<T>>
|
||||||
extends FrameworkCounterGroup<T, Counter> implements Group {
|
extends FrameworkCounterGroup<T, Counter> {
|
||||||
|
|
||||||
// Mix the framework counter implmementation into the Counter interface
|
|
||||||
class FrameworkCounterImpl extends FrameworkCounter implements Counter {
|
|
||||||
|
|
||||||
|
// Mix the framework counter implementation into the Counter interface
|
||||||
|
class FrameworkCounterImpl extends FrameworkCounter {
|
||||||
FrameworkCounterImpl(T key) {
|
FrameworkCounterImpl(T key) {
|
||||||
super(key);
|
super(key);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public boolean contentEquals(Counter counter) {
|
|
||||||
return equals(counter);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public long getCounter() {
|
|
||||||
return getValue();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
FrameworkGroupImpl(Class<T> cls) {
|
FrameworkGroupImpl(Class<T> cls) {
|
||||||
super(cls);
|
super(cls);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public long getCounter(String counterName) {
|
|
||||||
return getCounterValue(this, counterName);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override @Deprecated
|
|
||||||
public Counter getCounter(int id, String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public Counter getCounterForName(String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Counter newCounter(T key) {
|
protected Counter newCounter(T key) {
|
||||||
return new FrameworkCounterImpl(key);
|
return new Counter(new FrameworkCounterImpl(key));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return this;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Mix the file system counter group implementation into the Group interface
|
// Mix the file system counter group implementation into the Group interface
|
||||||
private static class FSGroupImpl extends FileSystemCounterGroup<Counter>
|
private static class FSGroupImpl extends FileSystemCounterGroup<Counter> {
|
||||||
implements Group {
|
|
||||||
|
|
||||||
private class FSCounterImpl extends FSCounter implements Counter {
|
private class FSCounterImpl extends FSCounter {
|
||||||
|
|
||||||
FSCounterImpl(String scheme, FileSystemCounter key) {
|
FSCounterImpl(String scheme, FileSystemCounter key) {
|
||||||
super(scheme, key);
|
super(scheme, key);
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override @Deprecated
|
|
||||||
public boolean contentEquals(Counter counter) {
|
|
||||||
throw new UnsupportedOperationException("Not supported yet.");
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public long getCounter() {
|
|
||||||
return getValue();
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Counter newCounter(String scheme, FileSystemCounter key) {
|
protected Counter newCounter(String scheme, FileSystemCounter key) {
|
||||||
return new FSCounterImpl(scheme, key);
|
return new Counter(new FSCounterImpl(scheme, key));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public long getCounter(String counterName) {
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
return getCounterValue(this, counterName);
|
return this;
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
|
||||||
public String makeEscapedCompactString() {
|
|
||||||
return toEscapedCompactString(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override @Deprecated
|
|
||||||
public Counter getCounter(int id, String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public Counter getCounterForName(String name) {
|
|
||||||
return findCounter(name);
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public synchronized Counter findCounter(String group, String name) {
|
public synchronized Counter findCounter(String group, String name) {
|
||||||
|
@ -342,7 +450,7 @@ public class Counters
|
||||||
FrameworkGroupFactory<Group> newFrameworkGroupFactory(final Class<T> cls) {
|
FrameworkGroupFactory<Group> newFrameworkGroupFactory(final Class<T> cls) {
|
||||||
return new FrameworkGroupFactory<Group>() {
|
return new FrameworkGroupFactory<Group>() {
|
||||||
@Override public Group newGroup(String name) {
|
@Override public Group newGroup(String name) {
|
||||||
return new FrameworkGroupImpl<T>(cls); // impl in this package
|
return new Group(new FrameworkGroupImpl<T>(cls)); // impl in this package
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
@ -350,12 +458,12 @@ public class Counters
|
||||||
@Override
|
@Override
|
||||||
protected Group newGenericGroup(String name, String displayName,
|
protected Group newGenericGroup(String name, String displayName,
|
||||||
Limits limits) {
|
Limits limits) {
|
||||||
return new GenericGroup(name, displayName, limits);
|
return new Group(new GenericGroup(name, displayName, limits));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
protected Group newFileSystemGroup() {
|
protected Group newFileSystemGroup() {
|
||||||
return new FSGroupImpl();
|
return new Group(new FSGroupImpl());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -72,4 +72,10 @@ public interface Counter extends Writable {
|
||||||
* @param incr the value to increase this counter by
|
* @param incr the value to increase this counter by
|
||||||
*/
|
*/
|
||||||
void increment(long incr);
|
void increment(long incr);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Return the underlying object if this is a facade.
|
||||||
|
* @return the undelying object.
|
||||||
|
*/
|
||||||
|
Counter getUnderlyingCounter();
|
||||||
}
|
}
|
||||||
|
|
|
@ -52,6 +52,11 @@ public class Counters extends AbstractCounters<Counter, CounterGroup> {
|
||||||
protected FrameworkCounter newCounter(T key) {
|
protected FrameworkCounter newCounter(T key) {
|
||||||
return new FrameworkCounter(key);
|
return new FrameworkCounter(key);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Mix generic group implementation into CounterGroup interface
|
// Mix generic group implementation into CounterGroup interface
|
||||||
|
@ -72,6 +77,11 @@ public class Counters extends AbstractCounters<Counter, CounterGroup> {
|
||||||
protected Counter newCounter() {
|
protected Counter newCounter() {
|
||||||
return new GenericCounter();
|
return new GenericCounter();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Mix file system group implementation into the CounterGroup interface
|
// Mix file system group implementation into the CounterGroup interface
|
||||||
|
@ -82,6 +92,11 @@ public class Counters extends AbstractCounters<Counter, CounterGroup> {
|
||||||
protected Counter newCounter(String scheme, FileSystemCounter key) {
|
protected Counter newCounter(String scheme, FileSystemCounter key) {
|
||||||
return new FSCounter(scheme, key);
|
return new FSCounter(scheme, key);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public CounterGroupBase<Counter> getUnderlyingGroup() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -172,7 +172,8 @@ public abstract class AbstractCounters<C extends Counter,
|
||||||
@InterfaceAudience.Private
|
@InterfaceAudience.Private
|
||||||
public synchronized C findCounter(String scheme, FileSystemCounter key) {
|
public synchronized C findCounter(String scheme, FileSystemCounter key) {
|
||||||
return ((FileSystemCounterGroup<C>) getGroup(
|
return ((FileSystemCounterGroup<C>) getGroup(
|
||||||
FileSystemCounter.class.getName())).findCounter(scheme, key);
|
FileSystemCounter.class.getName()).getUnderlyingGroup()).
|
||||||
|
findCounter(scheme, key);
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -243,11 +244,11 @@ public abstract class AbstractCounters<C extends Counter,
|
||||||
WritableUtils.writeVInt(out, groupFactory.version());
|
WritableUtils.writeVInt(out, groupFactory.version());
|
||||||
WritableUtils.writeVInt(out, fgroups.size()); // framework groups first
|
WritableUtils.writeVInt(out, fgroups.size()); // framework groups first
|
||||||
for (G group : fgroups.values()) {
|
for (G group : fgroups.values()) {
|
||||||
if (group instanceof FrameworkCounterGroup<?, ?>) {
|
if (group.getUnderlyingGroup() instanceof FrameworkCounterGroup<?, ?>) {
|
||||||
WritableUtils.writeVInt(out, GroupType.FRAMEWORK.ordinal());
|
WritableUtils.writeVInt(out, GroupType.FRAMEWORK.ordinal());
|
||||||
WritableUtils.writeVInt(out, getFrameworkGroupId(group.getName()));
|
WritableUtils.writeVInt(out, getFrameworkGroupId(group.getName()));
|
||||||
group.write(out);
|
group.write(out);
|
||||||
} else if (group instanceof FileSystemCounterGroup<?>) {
|
} else if (group.getUnderlyingGroup() instanceof FileSystemCounterGroup<?>) {
|
||||||
WritableUtils.writeVInt(out, GroupType.FILESYSTEM.ordinal());
|
WritableUtils.writeVInt(out, GroupType.FILESYSTEM.ordinal());
|
||||||
group.write(out);
|
group.write(out);
|
||||||
}
|
}
|
||||||
|
|
|
@ -98,4 +98,10 @@ public interface CounterGroupBase<T extends Counter>
|
||||||
* @param rightGroup the group to be added to this group
|
* @param rightGroup the group to be added to this group
|
||||||
*/
|
*/
|
||||||
void incrAllCounters(CounterGroupBase<T> rightGroup);
|
void incrAllCounters(CounterGroupBase<T> rightGroup);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Exposes the underlying group type if a facade.
|
||||||
|
* @return the underlying object that this object is wrapping up.
|
||||||
|
*/
|
||||||
|
CounterGroupBase<T> getUnderlyingGroup();
|
||||||
}
|
}
|
||||||
|
|
|
@ -110,6 +110,11 @@ public abstract class FileSystemCounterGroup<C extends Counter>
|
||||||
public void readFields(DataInput in) throws IOException {
|
public void readFields(DataInput in) throws IOException {
|
||||||
assert false : "shouldn't be called";
|
assert false : "shouldn't be called";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter getUnderlyingCounter() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
@ -231,10 +236,10 @@ public abstract class FileSystemCounterGroup<C extends Counter>
|
||||||
@Override
|
@Override
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
public void incrAllCounters(CounterGroupBase<C> other) {
|
public void incrAllCounters(CounterGroupBase<C> other) {
|
||||||
if (checkNotNull(other, "other group")
|
if (checkNotNull(other.getUnderlyingGroup(), "other group")
|
||||||
instanceof FileSystemCounterGroup<?>) {
|
instanceof FileSystemCounterGroup<?>) {
|
||||||
for (Counter counter : other) {
|
for (Counter counter : other) {
|
||||||
FSCounter c = (FSCounter) counter;
|
FSCounter c = (FSCounter) ((Counter)counter).getUnderlyingCounter();
|
||||||
findCounter(c.scheme, c.key) .increment(counter.getValue());
|
findCounter(c.scheme, c.key) .increment(counter.getValue());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -253,7 +258,7 @@ public abstract class FileSystemCounterGroup<C extends Counter>
|
||||||
for (Object counter : entry.getValue()) {
|
for (Object counter : entry.getValue()) {
|
||||||
if (counter == null) continue;
|
if (counter == null) continue;
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
FSCounter c = (FSCounter) counter;
|
FSCounter c = (FSCounter) ((Counter)counter).getUnderlyingCounter();
|
||||||
WritableUtils.writeVInt(out, c.key.ordinal()); // key
|
WritableUtils.writeVInt(out, c.key.ordinal()); // key
|
||||||
WritableUtils.writeVLong(out, c.getValue()); // value
|
WritableUtils.writeVLong(out, c.getValue()); // value
|
||||||
}
|
}
|
||||||
|
|
|
@ -18,21 +18,24 @@
|
||||||
|
|
||||||
package org.apache.hadoop.mapreduce.counters;
|
package org.apache.hadoop.mapreduce.counters;
|
||||||
|
|
||||||
|
import static com.google.common.base.Preconditions.checkNotNull;
|
||||||
|
|
||||||
import java.io.DataInput;
|
import java.io.DataInput;
|
||||||
import java.io.DataOutput;
|
import java.io.DataOutput;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.Arrays;
|
import java.util.Arrays;
|
||||||
import java.util.Iterator;
|
import java.util.Iterator;
|
||||||
|
|
||||||
import static com.google.common.base.Preconditions.*;
|
import org.apache.commons.logging.Log;
|
||||||
import com.google.common.collect.AbstractIterator;
|
import org.apache.commons.logging.LogFactory;
|
||||||
import com.google.common.collect.Iterators;
|
|
||||||
|
|
||||||
import org.apache.hadoop.classification.InterfaceAudience;
|
import org.apache.hadoop.classification.InterfaceAudience;
|
||||||
import org.apache.hadoop.io.WritableUtils;
|
import org.apache.hadoop.io.WritableUtils;
|
||||||
import org.apache.hadoop.mapreduce.Counter;
|
import org.apache.hadoop.mapreduce.Counter;
|
||||||
import org.apache.hadoop.mapreduce.util.ResourceBundles;
|
import org.apache.hadoop.mapreduce.util.ResourceBundles;
|
||||||
|
|
||||||
|
import com.google.common.collect.AbstractIterator;
|
||||||
|
import com.google.common.collect.Iterators;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* An abstract class to provide common implementation for the framework
|
* An abstract class to provide common implementation for the framework
|
||||||
* counter group in both mapred and mapreduce packages.
|
* counter group in both mapred and mapreduce packages.
|
||||||
|
@ -43,6 +46,7 @@ import org.apache.hadoop.mapreduce.util.ResourceBundles;
|
||||||
@InterfaceAudience.Private
|
@InterfaceAudience.Private
|
||||||
public abstract class FrameworkCounterGroup<T extends Enum<T>,
|
public abstract class FrameworkCounterGroup<T extends Enum<T>,
|
||||||
C extends Counter> implements CounterGroupBase<C> {
|
C extends Counter> implements CounterGroupBase<C> {
|
||||||
|
private static final Log LOG = LogFactory.getLog(FrameworkCounterGroup.class);
|
||||||
|
|
||||||
private final Class<T> enumClass; // for Enum.valueOf
|
private final Class<T> enumClass; // for Enum.valueOf
|
||||||
private final Object[] counters; // local casts are OK and save a class ref
|
private final Object[] counters; // local casts are OK and save a class ref
|
||||||
|
@ -95,6 +99,11 @@ public abstract class FrameworkCounterGroup<T extends Enum<T>,
|
||||||
public void readFields(DataInput in) throws IOException {
|
public void readFields(DataInput in) throws IOException {
|
||||||
assert false : "shouldn't be called";
|
assert false : "shouldn't be called";
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter getUnderlyingCounter() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@SuppressWarnings("unchecked")
|
@SuppressWarnings("unchecked")
|
||||||
|
|
|
@ -25,6 +25,7 @@ import java.io.IOException;
|
||||||
import org.apache.hadoop.classification.InterfaceAudience;
|
import org.apache.hadoop.classification.InterfaceAudience;
|
||||||
import org.apache.hadoop.io.Text;
|
import org.apache.hadoop.io.Text;
|
||||||
import org.apache.hadoop.io.WritableUtils;
|
import org.apache.hadoop.io.WritableUtils;
|
||||||
|
import org.apache.hadoop.mapreduce.Counter;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A generic counter implementation
|
* A generic counter implementation
|
||||||
|
@ -101,4 +102,9 @@ public class GenericCounter extends AbstractCounter {
|
||||||
public synchronized void increment(long incr) {
|
public synchronized void increment(long incr) {
|
||||||
value += incr;
|
value += incr;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public Counter getUnderlyingCounter() {
|
||||||
|
return this;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -24,6 +24,8 @@ import java.text.ParseException;
|
||||||
import java.util.Iterator;
|
import java.util.Iterator;
|
||||||
import java.util.Random;
|
import java.util.Random;
|
||||||
|
|
||||||
|
import org.apache.commons.logging.Log;
|
||||||
|
import org.apache.commons.logging.LogFactory;
|
||||||
import org.apache.hadoop.mapred.Counters.Counter;
|
import org.apache.hadoop.mapred.Counters.Counter;
|
||||||
import org.apache.hadoop.mapred.Counters.Group;
|
import org.apache.hadoop.mapred.Counters.Group;
|
||||||
import org.apache.hadoop.mapreduce.FileSystemCounter;
|
import org.apache.hadoop.mapreduce.FileSystemCounter;
|
||||||
|
@ -37,6 +39,7 @@ import org.junit.Test;
|
||||||
public class TestCounters {
|
public class TestCounters {
|
||||||
enum myCounters {TEST1, TEST2};
|
enum myCounters {TEST1, TEST2};
|
||||||
private static final long MAX_VALUE = 10;
|
private static final long MAX_VALUE = 10;
|
||||||
|
private static final Log LOG = LogFactory.getLog(TestCounters.class);
|
||||||
|
|
||||||
// Generates enum based counters
|
// Generates enum based counters
|
||||||
private Counters getEnumCounters(Enum[] keys) {
|
private Counters getEnumCounters(Enum[] keys) {
|
||||||
|
|
|
@ -18,6 +18,10 @@
|
||||||
|
|
||||||
package org.apache.hadoop.mapred;
|
package org.apache.hadoop.mapred;
|
||||||
|
|
||||||
|
import static org.junit.Assert.assertEquals;
|
||||||
|
import static org.junit.Assert.assertNotNull;
|
||||||
|
import static org.junit.Assert.assertTrue;
|
||||||
|
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.Formatter;
|
import java.util.Formatter;
|
||||||
|
@ -25,11 +29,6 @@ import java.util.Iterator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.StringTokenizer;
|
import java.util.StringTokenizer;
|
||||||
|
|
||||||
import org.junit.AfterClass;
|
|
||||||
import org.junit.BeforeClass;
|
|
||||||
import org.junit.Test;
|
|
||||||
import static org.junit.Assert.*;
|
|
||||||
|
|
||||||
import org.apache.commons.lang.RandomStringUtils;
|
import org.apache.commons.lang.RandomStringUtils;
|
||||||
import org.apache.hadoop.conf.Configuration;
|
import org.apache.hadoop.conf.Configuration;
|
||||||
import org.apache.hadoop.fs.FSDataOutputStream;
|
import org.apache.hadoop.fs.FSDataOutputStream;
|
||||||
|
@ -40,12 +39,14 @@ import org.apache.hadoop.io.LongWritable;
|
||||||
import org.apache.hadoop.io.Text;
|
import org.apache.hadoop.io.Text;
|
||||||
import org.apache.hadoop.io.Writable;
|
import org.apache.hadoop.io.Writable;
|
||||||
import org.apache.hadoop.io.WritableComparable;
|
import org.apache.hadoop.io.WritableComparable;
|
||||||
import org.apache.hadoop.mapreduce.Cluster;
|
|
||||||
import org.apache.hadoop.mapreduce.Job;
|
import org.apache.hadoop.mapreduce.Job;
|
||||||
import org.apache.hadoop.mapreduce.TaskCounter;
|
import org.apache.hadoop.mapreduce.TaskCounter;
|
||||||
import org.apache.hadoop.mapreduce.TaskType;
|
import org.apache.hadoop.mapreduce.TaskType;
|
||||||
import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
|
import org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter;
|
||||||
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter;
|
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormatCounter;
|
||||||
|
import org.junit.AfterClass;
|
||||||
|
import org.junit.BeforeClass;
|
||||||
|
import org.junit.Test;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This is an wordcount application that tests the count of records
|
* This is an wordcount application that tests the count of records
|
||||||
|
|
Loading…
Reference in New Issue