You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2013/09/25 09:31:31 UTC
[24/50] [abbrv] TEZ-443. Merge tez-dag-api and tez-engine-api into a
single module - tez-api (part of TEZ-398). (sseth)
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/AbstractCounters.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/AbstractCounters.java b/tez-api/src/main/java/org/apache/tez/common/counters/AbstractCounters.java
new file mode 100644
index 0000000..fd4fdee
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/AbstractCounters.java
@@ -0,0 +1,385 @@
+/*
+ * 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.common.counters;
+
+import static org.apache.tez.common.counters.CounterGroupFactory.getFrameworkGroupId;
+import static org.apache.tez.common.counters.CounterGroupFactory.isFrameworkGroup;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+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.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+
+/**
+ * An abstract class to provide common implementation for the Counters
+ * container in both mapred and mapreduce packages.
+ *
+ * @param <C> type of counter inside the counters
+ * @param <G> type of group inside the counters
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public abstract class AbstractCounters<C extends TezCounter,
+ G extends CounterGroupBase<C>>
+ implements Writable, Iterable<G> {
+
+ protected static final Log LOG = LogFactory.getLog("mapreduce.Counters");
+
+ /**
+ * A cache from enum values to the associated counter.
+ */
+ private Map<Enum<?>, C> cache = Maps.newIdentityHashMap();
+ //framework & fs groups
+ private Map<String, G> fgroups = new ConcurrentSkipListMap<String, G>();
+ // other groups
+ private Map<String, G> groups = new ConcurrentSkipListMap<String, G>();
+ private final CounterGroupFactory<C, G> groupFactory;
+
+ // For framework counter serialization without strings
+ enum GroupType { FRAMEWORK, FILESYSTEM };
+
+ // Writes only framework and fs counters if false.
+ private boolean writeAllCounters = true;
+
+ private static final Map<String, String> legacyMap = Maps.newHashMap();
+ static {
+ legacyMap.put("org.apache.hadoop.mapred.Task$Counter",
+ TaskCounter.class.getName());
+ legacyMap.put("org.apache.hadoop.mapred.JobInProgress$Counter",
+ JobCounter.class.getName());
+ legacyMap.put("FileSystemCounters", FileSystemCounter.class.getName());
+ }
+
+ private final Limits limits = new Limits();
+
+ @InterfaceAudience.Private
+ public AbstractCounters(CounterGroupFactory<C, G> gf) {
+ groupFactory = gf;
+ }
+
+ /**
+ * Construct from another counters object.
+ * @param <C1> type of the other counter
+ * @param <G1> type of the other counter group
+ * @param counters the counters object to copy
+ * @param groupFactory the factory for new groups
+ */
+ @InterfaceAudience.Private
+ public <C1 extends TezCounter, G1 extends CounterGroupBase<C1>>
+ AbstractCounters(AbstractCounters<C1, G1> counters,
+ CounterGroupFactory<C, G> groupFactory) {
+ this.groupFactory = groupFactory;
+ for(G1 group: counters) {
+ String name = group.getName();
+ G newGroup = groupFactory.newGroup(name, group.getDisplayName(), limits);
+ (isFrameworkGroup(name) ? fgroups : groups).put(name, newGroup);
+ for(TezCounter counter: group) {
+ newGroup.addCounter(counter.getName(), counter.getDisplayName(),
+ counter.getValue());
+ }
+ }
+ }
+
+ /** Add a group.
+ * @param group object to add
+ * @return the group
+ */
+ @InterfaceAudience.Private
+ public synchronized G addGroup(G group) {
+ String name = group.getName();
+ if (isFrameworkGroup(name)) {
+ fgroups.put(name, group);
+ } else {
+ limits.checkGroups(groups.size() + 1);
+ groups.put(name, group);
+ }
+ return group;
+ }
+
+ /**
+ * Add a new group
+ * @param name of the group
+ * @param displayName of the group
+ * @return the group
+ */
+ @InterfaceAudience.Private
+ public G addGroup(String name, String displayName) {
+ return addGroup(groupFactory.newGroup(name, displayName, limits));
+ }
+
+ /**
+ * Find a counter, create one if necessary
+ * @param groupName of the counter
+ * @param counterName name of the counter
+ * @return the matching counter
+ */
+ public C findCounter(String groupName, String counterName) {
+ G grp = getGroup(groupName);
+ return grp.findCounter(counterName);
+ }
+
+ /**
+ * Find the counter for the given enum. The same enum will always return the
+ * same counter.
+ * @param key the counter key
+ * @return the matching counter object
+ */
+ public synchronized C findCounter(Enum<?> key) {
+ C counter = cache.get(key);
+ if (counter == null) {
+ counter = findCounter(key.getDeclaringClass().getName(), key.name());
+ cache.put(key, counter);
+ }
+ return counter;
+ }
+
+ /**
+ * Find the file system counter for the given scheme and enum.
+ * @param scheme of the file system
+ * @param key the enum of the counter
+ * @return the file system counter
+ */
+ @InterfaceAudience.Private
+ public synchronized C findCounter(String scheme, FileSystemCounter key) {
+ return ((FileSystemCounterGroup<C>) getGroup(
+ FileSystemCounter.class.getName()).getUnderlyingGroup()).
+ findCounter(scheme, key);
+ }
+
+ /**
+ * Returns the names of all counter classes.
+ * @return Set of counter names.
+ */
+ public synchronized Iterable<String> getGroupNames() {
+ HashSet<String> deprecated = new HashSet<String>();
+ for(Map.Entry<String, String> entry : legacyMap.entrySet()) {
+ String newGroup = entry.getValue();
+ boolean isFGroup = isFrameworkGroup(newGroup);
+ if(isFGroup ? fgroups.containsKey(newGroup) : groups.containsKey(newGroup)) {
+ deprecated.add(entry.getKey());
+ }
+ }
+ return Iterables.concat(fgroups.keySet(), groups.keySet(), deprecated);
+ }
+
+ @Override
+ public Iterator<G> iterator() {
+ return Iterators.concat(fgroups.values().iterator(),
+ groups.values().iterator());
+ }
+
+ /**
+ * Returns the named counter group, or an empty group if there is none
+ * with the specified name.
+ * @param groupName name of the group
+ * @return the group
+ */
+ public synchronized G getGroup(String groupName) {
+
+ // filterGroupName
+ boolean groupNameInLegacyMap = true;
+ String newGroupName = legacyMap.get(groupName);
+ if (newGroupName == null) {
+ groupNameInLegacyMap = false;
+ newGroupName = Limits.filterGroupName(groupName);
+ }
+
+ boolean isFGroup = isFrameworkGroup(newGroupName);
+ G group = isFGroup ? fgroups.get(newGroupName) : groups.get(newGroupName);
+ if (group == null) {
+ group = groupFactory.newGroup(newGroupName, limits);
+ if (isFGroup) {
+ fgroups.put(newGroupName, group);
+ } else {
+ limits.checkGroups(groups.size() + 1);
+ groups.put(newGroupName, group);
+ }
+ if (groupNameInLegacyMap) {
+ LOG.warn("Group " + groupName + " is deprecated. Use " + newGroupName
+ + " instead");
+ }
+ }
+ return group;
+ }
+
+ /**
+ * Returns the total number of counters, by summing the number of counters
+ * in each group.
+ * @return the total number of counters
+ */
+ public synchronized int countCounters() {
+ int result = 0;
+ for (G group : this) {
+ result += group.size();
+ }
+ return result;
+ }
+
+ /**
+ * Write the set of groups.
+ * Counters ::= version #fgroups (groupId, group)* #groups (group)*
+ */
+ @Override
+ public synchronized void write(DataOutput out) throws IOException {
+ WritableUtils.writeVInt(out, groupFactory.version());
+ WritableUtils.writeVInt(out, fgroups.size()); // framework groups first
+ for (G group : fgroups.values()) {
+ if (group.getUnderlyingGroup() instanceof FrameworkCounterGroup<?, ?>) {
+ WritableUtils.writeVInt(out, GroupType.FRAMEWORK.ordinal());
+ WritableUtils.writeVInt(out, getFrameworkGroupId(group.getName()));
+ group.write(out);
+ } else if (group.getUnderlyingGroup() instanceof FileSystemCounterGroup<?>) {
+ WritableUtils.writeVInt(out, GroupType.FILESYSTEM.ordinal());
+ group.write(out);
+ }
+ }
+ if (writeAllCounters) {
+ WritableUtils.writeVInt(out, groups.size());
+ for (G group : groups.values()) {
+ Text.writeString(out, group.getName());
+ group.write(out);
+ }
+ } else {
+ WritableUtils.writeVInt(out, 0);
+ }
+ }
+
+ @Override
+ public synchronized void readFields(DataInput in) throws IOException {
+ int version = WritableUtils.readVInt(in);
+ if (version != groupFactory.version()) {
+ throw new IOException("Counters version mismatch, expected "+
+ groupFactory.version() +" got "+ version);
+ }
+ int numFGroups = WritableUtils.readVInt(in);
+ fgroups.clear();
+ GroupType[] groupTypes = GroupType.values();
+ while (numFGroups-- > 0) {
+ GroupType groupType = groupTypes[WritableUtils.readVInt(in)];
+ G group;
+ switch (groupType) {
+ case FILESYSTEM: // with nothing
+ group = groupFactory.newFileSystemGroup();
+ break;
+ case FRAMEWORK: // with group id
+ group = groupFactory.newFrameworkGroup(WritableUtils.readVInt(in));
+ break;
+ default: // Silence dumb compiler, as it would've thrown earlier
+ throw new IOException("Unexpected counter group type: "+ groupType);
+ }
+ group.readFields(in);
+ fgroups.put(group.getName(), group);
+ }
+ int numGroups = WritableUtils.readVInt(in);
+ while (numGroups-- > 0) {
+ limits.checkGroups(groups.size() + 1);
+ G group = groupFactory.newGenericGroup(Text.readString(in), null, limits);
+ group.readFields(in);
+ groups.put(group.getName(), group);
+ }
+ }
+
+ /**
+ * Return textual representation of the counter values.
+ * @return the string
+ */
+ @Override
+ public synchronized String toString() {
+ StringBuilder sb = new StringBuilder("Counters: " + countCounters());
+ for (G group: this) {
+ sb.append("\n\t").append(group.getDisplayName());
+ for (TezCounter counter: group) {
+ sb.append("\n\t\t").append(counter.getDisplayName()).append("=")
+ .append(counter.getValue());
+ }
+ }
+ return sb.toString();
+ }
+
+ /**
+ * Increments multiple counters by their amounts in another Counters
+ * instance.
+ * @param other the other Counters instance
+ */
+ public synchronized void incrAllCounters(AbstractCounters<C, G> other) {
+ for(G right : other) {
+ String groupName = right.getName();
+ G left = (isFrameworkGroup(groupName) ? fgroups : groups).get(groupName);
+ if (left == null) {
+ left = addGroup(groupName, right.getDisplayName());
+ }
+ left.incrAllCounters(right);
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean equals(Object genericRight) {
+ if (genericRight instanceof AbstractCounters<?, ?>) {
+ return Iterators.elementsEqual(iterator(),
+ ((AbstractCounters<C, G>)genericRight).iterator());
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ return groups.hashCode();
+ }
+
+ /**
+ * Set the "writeAllCounters" option to true or false
+ * @param send if true all counters would be serialized, otherwise only
+ * framework counters would be serialized in
+ * {@link #write(DataOutput)}
+ */
+ @InterfaceAudience.Private
+ public void setWriteAllCounters(boolean send) {
+ writeAllCounters = send;
+ }
+
+ /**
+ * Get the "writeAllCounters" option
+ * @return true of all counters would serialized
+ */
+ @InterfaceAudience.Private
+ public boolean getWriteAllCounters() {
+ return writeAllCounters;
+ }
+
+ @InterfaceAudience.Private
+ public Limits limits() {
+ return limits;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroup.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroup.java b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroup.java
new file mode 100644
index 0000000..bc7986d
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroup.java
@@ -0,0 +1,32 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * A group of {@link TezCounter}s that logically belong together. Typically,
+ * it is an {@link Enum} subclass and the counters are the values.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public interface CounterGroup extends CounterGroupBase<TezCounter> {
+ // essentially a typedef so user doesn't have to use generic syntax
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupBase.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupBase.java b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupBase.java
new file mode 100644
index 0000000..3b702ba
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupBase.java
@@ -0,0 +1,108 @@
+/*
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * The common counter group interface.
+ *
+ * @param <T> type of the counter for the group
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface CounterGroupBase<T extends TezCounter>
+ extends Writable, Iterable<T> {
+
+ /**
+ * Get the internal name of the group
+ * @return the internal name
+ */
+ String getName();
+
+ /**
+ * Get the display name of the group.
+ * @return the human readable name
+ */
+ String getDisplayName();
+
+ /**
+ * Set the display name of the group
+ * @param displayName of the group
+ */
+ void setDisplayName(String displayName);
+
+ /** Add a counter to this group.
+ * @param counter to add
+ */
+ void addCounter(T counter);
+
+ /**
+ * Add a counter to this group
+ * @param name of the counter
+ * @param displayName of the counter
+ * @param value of the counter
+ * @return the counter
+ */
+ T addCounter(String name, String displayName, long value);
+
+ /**
+ * Find a counter in the group.
+ * @param counterName the name of the counter
+ * @param displayName the display name of the counter
+ * @return the counter that was found or added
+ */
+ T findCounter(String counterName, String displayName);
+
+ /**
+ * Find a counter in the group
+ * @param counterName the name of the counter
+ * @param create create the counter if not found if true
+ * @return the counter that was found or added or null if create is false
+ */
+ T findCounter(String counterName, boolean create);
+
+ /**
+ * Find a counter in the group.
+ * @param counterName the name of the counter
+ * @return the counter that was found or added
+ */
+ T findCounter(String counterName);
+
+ /**
+ * @return the number of counters in this group.
+ */
+ int size();
+
+ /**
+ * Increment all counters by a group of counters
+ * @param rightGroup the group to be added to this group
+ */
+ void incrAllCounters(CounterGroupBase<T> rightGroup);
+
+ @Private
+ /**
+ * Exposes the underlying group type if a facade.
+ * @return the underlying object that this object is wrapping up.
+ */
+ CounterGroupBase<T> getUnderlyingGroup();
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupFactory.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupFactory.java b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupFactory.java
new file mode 100644
index 0000000..45da0dd
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/CounterGroupFactory.java
@@ -0,0 +1,180 @@
+/*
+ * 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.common.counters;
+
+import java.util.List;
+import java.util.Map;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * An abstract class to provide common implementation of the
+ * group factory in both mapred and mapreduce packages.
+ *
+ * @param <C> type of the counter
+ * @param <G> type of the group
+ */
+@InterfaceAudience.Private
+public abstract class CounterGroupFactory<C extends TezCounter,
+ G extends CounterGroupBase<C>> {
+
+ public interface FrameworkGroupFactory<F> {
+ F newGroup(String name);
+ }
+
+ // Integer mapping (for serialization) for framework groups
+ private static final Map<String, Integer> s2i = Maps.newHashMap();
+ private static final List<String> i2s = Lists.newArrayList();
+ private static final int VERSION = 1;
+ private static final String FS_GROUP_NAME = FileSystemCounter.class.getName();
+
+ private final Map<String, FrameworkGroupFactory<G>> fmap = Maps.newHashMap();
+ {
+ // Add builtin counter class here and the version when changed.
+ addFrameworkGroup(TaskCounter.class);
+ addFrameworkGroup(JobCounter.class);
+ addFrameworkGroup(DAGCounter.class);
+ }
+
+ // Initialize the framework counter group mapping
+ private synchronized <T extends Enum<T>>
+ void addFrameworkGroup(final Class<T> cls) {
+ updateFrameworkGroupMapping(cls);
+ fmap.put(cls.getName(), newFrameworkGroupFactory(cls));
+ }
+
+ // Update static mappings (c2i, i2s) of framework groups
+ private static synchronized void updateFrameworkGroupMapping(Class<?> cls) {
+ String name = cls.getName();
+ Integer i = s2i.get(name);
+ if (i != null) return;
+ i2s.add(name);
+ s2i.put(name, i2s.size() - 1);
+ }
+
+ /**
+ * Required override to return a new framework group factory
+ * @param <T> type of the counter enum class
+ * @param cls the counter enum class
+ * @return a new framework group factory
+ */
+ protected abstract <T extends Enum<T>>
+ FrameworkGroupFactory<G> newFrameworkGroupFactory(Class<T> cls);
+
+ /**
+ * Create a new counter group
+ * @param name of the group
+ * @param limits the counters limits policy object
+ * @return a new counter group
+ */
+ public G newGroup(String name, Limits limits) {
+ return newGroup(name, ResourceBundles.getCounterGroupName(name, name),
+ limits);
+ }
+
+ /**
+ * Create a new counter group
+ * @param name of the group
+ * @param displayName of the group
+ * @param limits the counters limits policy object
+ * @return a new counter group
+ */
+ public G newGroup(String name, String displayName, Limits limits) {
+ FrameworkGroupFactory<G> gf = fmap.get(name);
+ if (gf != null) return gf.newGroup(name);
+ if (name.equals(FS_GROUP_NAME)) {
+ return newFileSystemGroup();
+ } else if (s2i.get(name) != null) {
+ return newFrameworkGroup(s2i.get(name));
+ }
+ return newGenericGroup(name, displayName, limits);
+ }
+
+ /**
+ * Create a new framework group
+ * @param id of the group
+ * @return a new framework group
+ */
+ public G newFrameworkGroup(int id) {
+ String name;
+ synchronized(CounterGroupFactory.class) {
+ if (id < 0 || id >= i2s.size()) throwBadFrameGroupIdException(id);
+ name = i2s.get(id); // should not throw here.
+ }
+ FrameworkGroupFactory<G> gf = fmap.get(name);
+ if (gf == null) throwBadFrameGroupIdException(id);
+ return gf.newGroup(name);
+ }
+
+ /**
+ * Get the id of a framework group
+ * @param name of the group
+ * @return the framework group id
+ */
+ public static synchronized int getFrameworkGroupId(String name) {
+ Integer i = s2i.get(name);
+ if (i == null) throwBadFrameworkGroupNameException(name);
+ return i;
+ }
+
+ /**
+ * @return the counter factory version
+ */
+ public int version() {
+ return VERSION;
+ }
+
+ /**
+ * Check whether a group name is a name of a framework group (including
+ * the filesystem group).
+ *
+ * @param name to check
+ * @return true for framework group names
+ */
+ public static synchronized boolean isFrameworkGroup(String name) {
+ return s2i.get(name) != null || name.equals(FS_GROUP_NAME);
+ }
+
+ private static void throwBadFrameGroupIdException(int id) {
+ throw new IllegalArgumentException("bad framework group id: "+ id);
+ }
+
+ private static void throwBadFrameworkGroupNameException(String name) {
+ throw new IllegalArgumentException("bad framework group name: "+ name);
+ }
+
+ /**
+ * Abstract factory method to create a generic (vs framework) counter group
+ * @param name of the group
+ * @param displayName of the group
+ * @param limits limits of the counters
+ * @return a new generic counter group
+ */
+ protected abstract G newGenericGroup(String name, String displayName,
+ Limits limits);
+
+ /**
+ * Abstract factory method to create a file system counter group
+ * @return a new file system counter group
+ */
+ protected abstract G newFileSystemGroup();
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/DAGCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/DAGCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/DAGCounter.java
new file mode 100644
index 0000000..3598572
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/DAGCounter.java
@@ -0,0 +1,39 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+// Per-job counters
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public enum DAGCounter {
+ NUM_FAILED_TASKS,
+ NUM_KILLED_TASKS,
+ TOTAL_LAUNCHED_TASKS,
+ OTHER_LOCAL_TASKS,
+ DATA_LOCAL_TASKS,
+ RACK_LOCAL_TASKS,
+ SLOTS_MILLIS_TASKS,
+ FALLOW_SLOTS_MILLIS_TASKS,
+ TOTAL_LAUNCHED_UBERTASKS,
+ NUM_UBER_SUBTASKS,
+ NUM_FAILED_UBERTASKS
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounter.java
new file mode 100644
index 0000000..08f4c5d
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounter.java
@@ -0,0 +1,30 @@
+/*
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+@InterfaceAudience.Private
+public enum FileSystemCounter {
+ BYTES_READ,
+ BYTES_WRITTEN,
+ READ_OPS,
+ LARGE_READ_OPS,
+ WRITE_OPS,
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounterGroup.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounterGroup.java b/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounterGroup.java
new file mode 100644
index 0000000..d4b167a
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/FileSystemCounterGroup.java
@@ -0,0 +1,327 @@
+/*
+ * 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.common.counters;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.Iterator;
+import java.util.Locale;
+import java.util.Map;
+
+import com.google.common.base.Joiner;
+import static com.google.common.base.Preconditions.*;
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.io.WritableUtils;
+
+/**
+ * An abstract class to provide common implementation of the filesystem
+ * counter group in both mapred and mapreduce packages.
+ *
+ * @param <C> the type of the Counter for the group
+ */
+@InterfaceAudience.Private
+public abstract class FileSystemCounterGroup<C extends TezCounter>
+ implements CounterGroupBase<C> {
+
+ static final int MAX_NUM_SCHEMES = 100; // intern/sanity check
+ static final ConcurrentMap<String, String> schemes = Maps.newConcurrentMap();
+
+ // C[] would need Array.newInstance which requires a Class<C> reference.
+ // Just a few local casts probably worth not having to carry it around.
+ private final Map<String, Object[]> map =
+ new ConcurrentSkipListMap<String, Object[]>();
+ private String displayName;
+
+ private static final Joiner NAME_JOINER = Joiner.on('_');
+ private static final Joiner DISP_JOINER = Joiner.on(": ");
+
+ @InterfaceAudience.Private
+ public static class FSCounter extends AbstractCounter {
+ final String scheme;
+ final FileSystemCounter key;
+ private long value;
+
+ public FSCounter(String scheme, FileSystemCounter ref) {
+ this.scheme = scheme;
+ key = ref;
+ }
+
+ @Override
+ public String getName() {
+ return NAME_JOINER.join(scheme, key.name());
+ }
+
+ @Override
+ public String getDisplayName() {
+ return DISP_JOINER.join(scheme, localizeCounterName(key.name()));
+ }
+
+ protected String localizeCounterName(String counterName) {
+ return ResourceBundles.getCounterName(FileSystemCounter.class.getName(),
+ counterName, counterName);
+ }
+
+ @Override
+ public long getValue() {
+ return value;
+ }
+
+ @Override
+ public void setValue(long value) {
+ this.value = value;
+ }
+
+ @Override
+ public void increment(long incr) {
+ value += incr;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ assert false : "shouldn't be called";
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ assert false : "shouldn't be called";
+ }
+
+ @Override
+ public TezCounter getUnderlyingCounter() {
+ return this;
+ }
+ }
+
+ @Override
+ public String getName() {
+ return FileSystemCounter.class.getName();
+ }
+
+ @Override
+ public String getDisplayName() {
+ if (displayName == null) {
+ displayName = ResourceBundles.getCounterGroupName(getName(),
+ "File System Counters");
+ }
+ return displayName;
+ }
+
+ @Override
+ public void setDisplayName(String displayName) {
+ this.displayName = displayName;
+ }
+
+ @Override
+ public void addCounter(C counter) {
+ C ours;
+ if (counter instanceof FileSystemCounterGroup.FSCounter) {
+ FSCounter c = (FSCounter) counter;
+ ours = findCounter(c.scheme, c.key);
+ }
+ else {
+ ours = findCounter(counter.getName());
+ }
+ ours.setValue(counter.getValue());
+ }
+
+ @Override
+ public C addCounter(String name, String displayName, long value) {
+ C counter = findCounter(name);
+ counter.setValue(value);
+ return counter;
+ }
+
+ // Parse generic counter name into [scheme, key]
+ private String[] parseCounterName(String counterName) {
+ int schemeEnd = counterName.indexOf('_');
+ if (schemeEnd < 0) {
+ throw new IllegalArgumentException("bad fs counter name");
+ }
+ return new String[]{counterName.substring(0, schemeEnd),
+ counterName.substring(schemeEnd + 1)};
+ }
+
+ @Override
+ public C findCounter(String counterName, String displayName) {
+ return findCounter(counterName);
+ }
+
+ @Override
+ public C findCounter(String counterName, boolean create) {
+ try {
+ String[] pair = parseCounterName(counterName);
+ return findCounter(pair[0], FileSystemCounter.valueOf(pair[1]));
+ }
+ catch (Exception e) {
+ if (create) throw new IllegalArgumentException(e);
+ return null;
+ }
+ }
+
+ @Override
+ public C findCounter(String counterName) {
+ return findCounter(counterName, true);
+ }
+
+ @SuppressWarnings("unchecked")
+ public synchronized C findCounter(String scheme, FileSystemCounter key) {
+ final String canonicalScheme = checkScheme(scheme);
+ Object[] counters = map.get(canonicalScheme);
+ int ord = key.ordinal();
+ if (counters == null) {
+ counters = new Object[FileSystemCounter.values().length];
+ map.put(canonicalScheme, counters);
+ counters[ord] = newCounter(canonicalScheme, key);
+ }
+ else if (counters[ord] == null) {
+ counters[ord] = newCounter(canonicalScheme, key);
+ }
+ return (C) counters[ord];
+ }
+
+ private String checkScheme(String scheme) {
+ String fixed = scheme.toUpperCase(Locale.US);
+ String interned = schemes.putIfAbsent(fixed, fixed);
+ if (schemes.size() > MAX_NUM_SCHEMES) {
+ // mistakes or abuses
+ throw new IllegalArgumentException("too many schemes? "+ schemes.size() +
+ " when process scheme: "+ scheme);
+ }
+ return interned == null ? fixed : interned;
+ }
+
+ /**
+ * Abstract factory method to create a file system counter
+ * @param scheme of the file system
+ * @param key the enum of the file system counter
+ * @return a new file system counter
+ */
+ protected abstract C newCounter(String scheme, FileSystemCounter key);
+
+ @Override
+ public int size() {
+ int n = 0;
+ for (Object[] counters : map.values()) {
+ n += numSetCounters(counters);
+ }
+ return n;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void incrAllCounters(CounterGroupBase<C> other) {
+ if (checkNotNull(other.getUnderlyingGroup(), "other group")
+ instanceof FileSystemCounterGroup<?>) {
+ for (TezCounter counter : other) {
+ FSCounter c = (FSCounter) ((TezCounter)counter).getUnderlyingCounter();
+ findCounter(c.scheme, c.key) .increment(counter.getValue());
+ }
+ }
+ }
+
+ /**
+ * FileSystemGroup ::= #scheme (scheme #counter (key value)*)*
+ */
+ @Override
+ public void write(DataOutput out) throws IOException {
+ WritableUtils.writeVInt(out, map.size()); // #scheme
+ for (Map.Entry<String, Object[]> entry : map.entrySet()) {
+ WritableUtils.writeString(out, entry.getKey()); // scheme
+ // #counter for the above scheme
+ WritableUtils.writeVInt(out, numSetCounters(entry.getValue()));
+ for (Object counter : entry.getValue()) {
+ if (counter == null) continue;
+ @SuppressWarnings("unchecked")
+ FSCounter c = (FSCounter) ((TezCounter)counter).getUnderlyingCounter();
+ WritableUtils.writeVInt(out, c.key.ordinal()); // key
+ WritableUtils.writeVLong(out, c.getValue()); // value
+ }
+ }
+ }
+
+ private int numSetCounters(Object[] counters) {
+ int n = 0;
+ for (Object counter : counters) if (counter != null) ++n;
+ return n;
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ int numSchemes = WritableUtils.readVInt(in); // #scheme
+ FileSystemCounter[] enums = FileSystemCounter.values();
+ for (int i = 0; i < numSchemes; ++i) {
+ String scheme = WritableUtils.readString(in); // scheme
+ int numCounters = WritableUtils.readVInt(in); // #counter
+ for (int j = 0; j < numCounters; ++j) {
+ findCounter(scheme, enums[WritableUtils.readVInt(in)]) // key
+ .setValue(WritableUtils.readVLong(in)); // value
+ }
+ }
+ }
+
+ @Override
+ public Iterator<C> iterator() {
+ return new AbstractIterator<C>() {
+ Iterator<Object[]> it = map.values().iterator();
+ Object[] counters = it.hasNext() ? it.next() : null;
+ int i = 0;
+ @Override
+ protected C computeNext() {
+ while (counters != null) {
+ while (i < counters.length) {
+ @SuppressWarnings("unchecked")
+ C counter = (C) counters[i++];
+ if (counter != null) return counter;
+ }
+ i = 0;
+ counters = it.hasNext() ? it.next() : null;
+ }
+ return endOfData();
+ }
+ };
+ }
+
+ @Override
+ public synchronized boolean equals(Object genericRight) {
+ if (genericRight instanceof CounterGroupBase<?>) {
+ @SuppressWarnings("unchecked")
+ CounterGroupBase<C> right = (CounterGroupBase<C>) genericRight;
+ return Iterators.elementsEqual(iterator(), right.iterator());
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized int hashCode() {
+ // need to be deep as counters is an array
+ int hash = FileSystemCounter.class.hashCode();
+ for (Object[] counters : map.values()) {
+ if (counters != null) hash ^= Arrays.hashCode(counters);
+ }
+ return hash;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/FrameworkCounterGroup.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/FrameworkCounterGroup.java b/tez-api/src/main/java/org/apache/tez/common/counters/FrameworkCounterGroup.java
new file mode 100644
index 0000000..42fb636
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/FrameworkCounterGroup.java
@@ -0,0 +1,275 @@
+/*
+ * 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.common.counters;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Iterator;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.io.WritableUtils;
+
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Iterators;
+
+/**
+ * An abstract class to provide common implementation for the framework
+ * counter group in both mapred and mapreduce packages.
+ *
+ * @param <T> type of the counter enum class
+ * @param <C> type of the counter
+ */
+@InterfaceAudience.Private
+public abstract class FrameworkCounterGroup<T extends Enum<T>,
+ C extends TezCounter> implements CounterGroupBase<C> {
+ private static final Log LOG = LogFactory.getLog(FrameworkCounterGroup.class);
+
+ private final Class<T> enumClass; // for Enum.valueOf
+ private final Object[] counters; // local casts are OK and save a class ref
+ private String displayName = null;
+
+ /**
+ * A counter facade for framework counters.
+ * Use old (which extends new) interface to make compatibility easier.
+ */
+ @InterfaceAudience.Private
+ public static class FrameworkCounter<T extends Enum<T>> extends AbstractCounter {
+ final T key;
+ final String groupName;
+ private long value;
+
+ public FrameworkCounter(T ref, String groupName) {
+ key = ref;
+ this.groupName = groupName;
+ }
+
+ @Override
+ public String getName() {
+ return key.name();
+ }
+
+ @Override
+ public String getDisplayName() {
+ return ResourceBundles.getCounterName(groupName, getName(), getName());
+ }
+
+ @Override
+ public long getValue() {
+ return value;
+ }
+
+ @Override
+ public void setValue(long value) {
+ this.value = value;
+ }
+
+ @Override
+ public void increment(long incr) {
+ value += incr;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ assert false : "shouldn't be called";
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ assert false : "shouldn't be called";
+ }
+
+ @Override
+ public TezCounter getUnderlyingCounter() {
+ return this;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public FrameworkCounterGroup(Class<T> enumClass) {
+ this.enumClass = enumClass;
+ T[] enums = enumClass.getEnumConstants();
+ counters = new Object[enums.length];
+ }
+
+ @Override
+ public String getName() {
+ return enumClass.getName();
+ }
+
+ @Override
+ public String getDisplayName() {
+ if (displayName == null) {
+ displayName = ResourceBundles.getCounterGroupName(getName(), getName());
+ }
+ return displayName;
+ }
+
+ @Override
+ public void setDisplayName(String displayName) {
+ this.displayName = displayName;
+ }
+
+ private T valueOf(String name) {
+ return Enum.valueOf(enumClass, name);
+ }
+
+ @Override
+ public void addCounter(C counter) {
+ C ours = findCounter(counter.getName());
+ ours.setValue(counter.getValue());
+ }
+
+ @Override
+ public C addCounter(String name, String displayName, long value) {
+ C counter = findCounter(name);
+ counter.setValue(value);
+ return counter;
+ }
+
+ @Override
+ public C findCounter(String counterName, String displayName) {
+ return findCounter(counterName);
+ }
+
+ @Override
+ public C findCounter(String counterName, boolean create) {
+ try {
+ return findCounter(valueOf(counterName));
+ }
+ catch (Exception e) {
+ if (create) throw new IllegalArgumentException(e);
+ return null;
+ }
+ }
+
+ @Override
+ public C findCounter(String counterName) {
+ return findCounter(valueOf(counterName));
+ }
+
+ @SuppressWarnings("unchecked")
+ private C findCounter(T key) {
+ int i = key.ordinal();
+ if (counters[i] == null) {
+ counters[i] = newCounter(key);
+ }
+ return (C) counters[i];
+ }
+
+ /**
+ * Abstract factory method for new framework counter
+ * @param key for the enum value of a counter
+ * @return a new counter for the key
+ */
+ protected abstract C newCounter(T key);
+
+ @Override
+ public int size() {
+ int n = 0;
+ for (int i = 0; i < counters.length; ++i) {
+ if (counters[i] != null) ++n;
+ }
+ return n;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void incrAllCounters(CounterGroupBase<C> other) {
+ if (checkNotNull(other, "other counter group")
+ instanceof FrameworkCounterGroup<?, ?>) {
+ for (TezCounter counter : other) {
+ findCounter(((FrameworkCounter) counter).key.name())
+ .increment(counter.getValue());
+ }
+ }
+ }
+
+ /**
+ * FrameworkGroup ::= #counter (key value)*
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public void write(DataOutput out) throws IOException {
+ WritableUtils.writeVInt(out, size());
+ for (int i = 0; i < counters.length; ++i) {
+ TezCounter counter = (C) counters[i];
+ if (counter != null) {
+ WritableUtils.writeVInt(out, i);
+ WritableUtils.writeVLong(out, counter.getValue());
+ }
+ }
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ clear();
+ int len = WritableUtils.readVInt(in);
+ T[] enums = enumClass.getEnumConstants();
+ for (int i = 0; i < len; ++i) {
+ int ord = WritableUtils.readVInt(in);
+ TezCounter counter = newCounter(enums[ord]);
+ counter.setValue(WritableUtils.readVLong(in));
+ counters[ord] = counter;
+ }
+ }
+
+ private void clear() {
+ for (int i = 0; i < counters.length; ++i) {
+ counters[i] = null;
+ }
+ }
+
+ @Override
+ public Iterator<C> iterator() {
+ return new AbstractIterator<C>() {
+ int i = 0;
+ @Override
+ protected C computeNext() {
+ while (i < counters.length) {
+ @SuppressWarnings("unchecked")
+ C counter = (C) counters[i++];
+ if (counter != null) return counter;
+ }
+ return endOfData();
+ }
+ };
+ }
+
+ @Override
+ public boolean equals(Object genericRight) {
+ if (genericRight instanceof CounterGroupBase<?>) {
+ @SuppressWarnings("unchecked")
+ CounterGroupBase<C> right = (CounterGroupBase<C>) genericRight;
+ return Iterators.elementsEqual(iterator(), right.iterator());
+ }
+ return false;
+ }
+
+ @Override
+ public synchronized int hashCode() {
+ // need to be deep as counters is an array
+ return Arrays.deepHashCode(new Object[]{enumClass, counters, displayName});
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/GenericCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/GenericCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/GenericCounter.java
new file mode 100644
index 0000000..5477606
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/GenericCounter.java
@@ -0,0 +1,109 @@
+/*
+ * 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.common.counters;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
+
+/**
+ * A generic counter implementation
+ */
+@InterfaceAudience.Private
+public class GenericCounter extends AbstractCounter {
+
+ private String name;
+ private String displayName;
+ private long value = 0;
+
+ public GenericCounter() {
+ // mostly for readFields
+ }
+
+ public GenericCounter(String name, String displayName) {
+ this.name = name;
+ this.displayName = displayName;
+ }
+
+ public GenericCounter(String name, String displayName, long value) {
+ this.name = name;
+ this.displayName = displayName;
+ this.value = value;
+ }
+
+ @Override @Deprecated
+ public synchronized void setDisplayName(String displayName) {
+ this.displayName = displayName;
+ }
+
+ @Override
+ public synchronized void readFields(DataInput in) throws IOException {
+ name = Text.readString(in);
+ displayName = in.readBoolean() ? Text.readString(in) : name;
+ value = WritableUtils.readVLong(in);
+ }
+
+ /**
+ * GenericCounter ::= keyName isDistinctDisplayName [displayName] value
+ */
+ @Override
+ public synchronized void write(DataOutput out) throws IOException {
+ Text.writeString(out, name);
+ boolean distinctDisplayName = ! name.equals(displayName);
+ out.writeBoolean(distinctDisplayName);
+ if (distinctDisplayName) {
+ Text.writeString(out, displayName);
+ }
+ WritableUtils.writeVLong(out, value);
+ }
+
+ @Override
+ public synchronized String getName() {
+ return name;
+ }
+
+ @Override
+ public synchronized String getDisplayName() {
+ return displayName;
+ }
+
+ @Override
+ public synchronized long getValue() {
+ return value;
+ }
+
+ @Override
+ public synchronized void setValue(long value) {
+ this.value = value;
+ }
+
+ @Override
+ public synchronized void increment(long incr) {
+ value += incr;
+ }
+
+ @Override
+ public TezCounter getUnderlyingCounter() {
+ return this;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/JobCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/JobCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/JobCounter.java
new file mode 100644
index 0000000..1eb2be8
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/JobCounter.java
@@ -0,0 +1,45 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+// Per-job counters
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public enum JobCounter {
+ NUM_FAILED_MAPS,
+ NUM_FAILED_REDUCES,
+ NUM_KILLED_MAPS,
+ NUM_KILLED_REDUCES,
+ TOTAL_LAUNCHED_MAPS,
+ TOTAL_LAUNCHED_REDUCES,
+ OTHER_LOCAL_MAPS,
+ DATA_LOCAL_MAPS,
+ RACK_LOCAL_MAPS,
+ SLOTS_MILLIS_MAPS,
+ SLOTS_MILLIS_REDUCES,
+ FALLOW_SLOTS_MILLIS_MAPS,
+ FALLOW_SLOTS_MILLIS_REDUCES,
+ TOTAL_LAUNCHED_UBERTASKS,
+ NUM_UBER_SUBMAPS,
+ NUM_UBER_SUBREDUCES,
+ NUM_FAILED_UBERTASKS
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/LimitExceededException.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/LimitExceededException.java b/tez-api/src/main/java/org/apache/tez/common/counters/LimitExceededException.java
new file mode 100644
index 0000000..e50bd81
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/LimitExceededException.java
@@ -0,0 +1,36 @@
+/*
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class LimitExceededException extends RuntimeException {
+
+ private static final long serialVersionUID = 1L;
+
+ public LimitExceededException(String msg) {
+ super(msg);
+ }
+
+ // Only allows chaining of related exceptions
+ public LimitExceededException(LimitExceededException cause) {
+ super(cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/Limits.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/Limits.java b/tez-api/src/main/java/org/apache/tez/common/counters/Limits.java
new file mode 100644
index 0000000..aacce87
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/Limits.java
@@ -0,0 +1,112 @@
+/*
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.common.TezJobConfig;
+
+@InterfaceAudience.Private
+public class Limits {
+
+ private static Configuration conf = null;
+ private static int GROUP_NAME_MAX;
+ private static int COUNTER_NAME_MAX;
+ private static int GROUPS_MAX;
+ private static int COUNTERS_MAX;
+ private static boolean initialized = false;
+
+ private static synchronized void ensureInitialized() {
+ if (initialized) {
+ return;
+ }
+ if (conf == null) {
+ conf = new Configuration();
+ }
+ GROUP_NAME_MAX =
+ conf.getInt(TezJobConfig.COUNTER_GROUP_NAME_MAX_KEY,
+ TezJobConfig.COUNTER_GROUP_NAME_MAX_DEFAULT);
+ COUNTER_NAME_MAX =
+ conf.getInt(TezJobConfig.COUNTER_NAME_MAX_KEY,
+ TezJobConfig.COUNTER_NAME_MAX_DEFAULT);
+ GROUPS_MAX =
+ conf.getInt(TezJobConfig.COUNTER_GROUPS_MAX_KEY,
+ TezJobConfig.COUNTER_GROUPS_MAX_DEFAULT);
+ COUNTERS_MAX =
+ conf.getInt(TezJobConfig.COUNTERS_MAX_KEY, TezJobConfig.
+ COUNTERS_MAX_DEFAULT);
+ initialized = true;
+ }
+
+ private int totalCounters;
+ private LimitExceededException firstViolation;
+
+ public static String filterName(String name, int maxLen) {
+ return name.length() > maxLen ? name.substring(0, maxLen - 1) : name;
+ }
+
+ public static String filterCounterName(String name) {
+ ensureInitialized();
+ return filterName(name, COUNTER_NAME_MAX);
+ }
+
+ public static String filterGroupName(String name) {
+ ensureInitialized();
+ return filterName(name, GROUP_NAME_MAX);
+ }
+
+ public synchronized void checkCounters(int size) {
+ ensureInitialized();
+ if (firstViolation != null) {
+ throw new LimitExceededException(firstViolation);
+ }
+ if (size > COUNTERS_MAX) {
+ firstViolation = new LimitExceededException("Too many counters: "+ size +
+ " max="+ COUNTERS_MAX);
+ throw firstViolation;
+ }
+ }
+
+ public synchronized void incrCounters() {
+ checkCounters(totalCounters + 1);
+ ++totalCounters;
+ }
+
+ public synchronized void checkGroups(int size) {
+ ensureInitialized();
+ if (firstViolation != null) {
+ throw new LimitExceededException(firstViolation);
+ }
+ if (size > GROUPS_MAX) {
+ firstViolation = new LimitExceededException("Too many counter groups: "+
+ size +" max="+ GROUPS_MAX);
+ }
+ }
+
+ public synchronized LimitExceededException violation() {
+ return firstViolation;
+ }
+
+ public synchronized static void setConfiguration(Configuration conf) {
+ if (Limits.conf == null && conf != null) {
+ Limits.conf = conf;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/ResourceBundles.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/ResourceBundles.java b/tez-api/src/main/java/org/apache/tez/common/counters/ResourceBundles.java
new file mode 100644
index 0000000..8113cab
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/ResourceBundles.java
@@ -0,0 +1,94 @@
+/*
+ * 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.common.counters;
+
+import java.util.Locale;
+import java.util.ResourceBundle;
+import java.util.MissingResourceException;
+
+/**
+ * Helper class to handle resource bundles in a saner way
+ */
+public class ResourceBundles {
+
+ /**
+ * Get a resource bundle
+ * @param bundleName of the resource
+ * @return the resource bundle
+ * @throws MissingResourceException
+ */
+ public static ResourceBundle getBundle(String bundleName) {
+ return ResourceBundle.getBundle(bundleName.replace('$', '_'),
+ Locale.getDefault(), Thread.currentThread().getContextClassLoader());
+ }
+
+ /**
+ * Get a resource given bundle name and key
+ * @param <T> type of the resource
+ * @param bundleName name of the resource bundle
+ * @param key to lookup the resource
+ * @param suffix for the key to lookup
+ * @param defaultValue of the resource
+ * @return the resource or the defaultValue
+ * @throws ClassCastException if the resource found doesn't match T
+ */
+ @SuppressWarnings("unchecked")
+ public static synchronized <T> T getValue(String bundleName, String key,
+ String suffix, T defaultValue) {
+ T value;
+ try {
+ ResourceBundle bundle = getBundle(bundleName);
+ value = (T) bundle.getObject(getLookupKey(key, suffix));
+ if (value != null) {
+ return value;
+ }
+ }
+ catch (Exception e) {
+ // Ignore
+ }
+ return defaultValue;
+ }
+
+ private static String getLookupKey(String key, String suffix) {
+ if (suffix == null || suffix.isEmpty()) return key;
+ return key + suffix;
+ }
+
+ /**
+ * Get the counter group display name
+ * @param group the group name to lookup
+ * @param defaultValue of the group
+ * @return the group display name
+ */
+ public static String getCounterGroupName(String group, String defaultValue) {
+ return getValue(group, "CounterGroupName", "", defaultValue);
+ }
+
+ /**
+ * Get the counter display name
+ * @param group the counter group name for the counter
+ * @param counter the counter name to lookup
+ * @param defaultValue of the counter
+ * @return the counter display name
+ */
+ public static String getCounterName(String group, String counter,
+ String defaultValue) {
+ return getValue(group, counter, ".name", defaultValue);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java
new file mode 100644
index 0000000..b6fca27
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java
@@ -0,0 +1,66 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+// TODO TEZAM5 For MR compatibility, a conversion from tez.TaskCounters to
+// mapreduce.TaskCounters will likely be required somewhere.
+// Similarly for FileSystemCounters and others.
+
+// Counters used by Task classes
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public enum TaskCounter {
+ // TODO Eventually, rename counters to be non-MR specific and map them to MR equivalent.
+ MAP_INPUT_RECORDS,
+ MAP_OUTPUT_RECORDS,
+ MAP_SKIPPED_RECORDS,
+ MAP_OUTPUT_BYTES,
+ MAP_OUTPUT_MATERIALIZED_BYTES,
+ SPLIT_RAW_BYTES,
+ COMBINE_INPUT_RECORDS,
+ COMBINE_OUTPUT_RECORDS,
+ REDUCE_INPUT_GROUPS,
+ REDUCE_SHUFFLE_BYTES,
+ REDUCE_INPUT_RECORDS,
+ REDUCE_OUTPUT_RECORDS,
+ REDUCE_SKIPPED_GROUPS,
+ REDUCE_SKIPPED_RECORDS,
+ SPILLED_RECORDS,
+ SHUFFLED_MAPS,
+ FAILED_SHUFFLE,
+ MERGED_MAP_OUTPUTS,
+ GC_TIME_MILLIS,
+ CPU_MILLISECONDS,
+ PHYSICAL_MEMORY_BYTES,
+ VIRTUAL_MEMORY_BYTES,
+ COMMITTED_HEAP_BYTES,
+
+ INPUT_RECORDS,
+ OUTPUT_RECORDS,
+ SKIPPED_RECORDS,
+ OUTPUT_BYTES,
+ OUTPUT_MATERIALIZED_BYTES,
+ INPUT_GROUPS,
+ SHUFFLE_BYTES,
+ SHUFFLED_TASKS,
+ MERGED_TASK_OUTPUTS,
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/TezCounter.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/TezCounter.java b/tez-api/src/main/java/org/apache/tez/common/counters/TezCounter.java
new file mode 100644
index 0000000..394c820
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/TezCounter.java
@@ -0,0 +1,83 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.io.Writable;
+
+/**
+ * A named counter that tracks the progress of a map/reduce job.
+ *
+ * <p><code>Counters</code> represent global counters, defined either by the
+ * Map-Reduce framework or applications. Each <code>Counter</code> is named by
+ * an {@link Enum} and has a long for the value.</p>
+ *
+ * <p><code>Counters</code> are bunched into Groups, each comprising of
+ * counters from a particular <code>Enum</code> class.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public interface TezCounter extends Writable {
+
+ /**
+ * Set the display name of the counter
+ * @param displayName of the counter
+ * @deprecated (and no-op by default)
+ */
+ @Deprecated
+ void setDisplayName(String displayName);
+
+ /**
+ * @return the name of the counter
+ */
+ String getName();
+
+ /**
+ * Get the display name of the counter.
+ * @return the user facing name of the counter
+ */
+ String getDisplayName();
+
+ /**
+ * What is the current value of this counter?
+ * @return the current value
+ */
+ long getValue();
+
+ /**
+ * Set this counter by the given value
+ * @param value the value to set
+ */
+ void setValue(long value);
+
+ /**
+ * Increment this counter by the given value
+ * @param incr the value to increase this counter by
+ */
+ void increment(long incr);
+
+ /**
+ * Return the underlying object if this is a facade.
+ * @return the undelying object.
+ */
+ @Private
+ TezCounter getUnderlyingCounter();
+}
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/d316f723/tez-api/src/main/java/org/apache/tez/common/counters/TezCounters.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/common/counters/TezCounters.java b/tez-api/src/main/java/org/apache/tez/common/counters/TezCounters.java
new file mode 100644
index 0000000..1c9521a
--- /dev/null
+++ b/tez-api/src/main/java/org/apache/tez/common/counters/TezCounters.java
@@ -0,0 +1,144 @@
+/**
+ * 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.common.counters;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * <p><code>Counters</code> holds per job/task counters, defined either by the
+ * Map-Reduce framework or applications. Each <code>Counter</code> can be of
+ * any {@link Enum} type.</p>
+ *
+ * <p><code>Counters</code> are bunched into {@link CounterGroup}s, each
+ * comprising of counters from a particular <code>Enum</code> class.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Stable
+public class TezCounters extends AbstractCounters<TezCounter, CounterGroup> {
+
+ // Mix framework group implementation into CounterGroup interface
+ private static class FrameworkGroupImpl<T extends Enum<T>>
+ extends FrameworkCounterGroup<T, TezCounter> implements CounterGroup {
+
+ FrameworkGroupImpl(Class<T> cls) {
+ super(cls);
+ }
+
+ @Override
+ protected FrameworkCounter<T> newCounter(T key) {
+ return new FrameworkCounter<T>(key, getName());
+ }
+
+ @Override
+ public CounterGroupBase<TezCounter> getUnderlyingGroup() {
+ return this;
+ }
+ }
+
+ // Mix generic group implementation into CounterGroup interface
+ // and provide some mandatory group factory methods.
+ private static class GenericGroup extends AbstractCounterGroup<TezCounter>
+ implements CounterGroup {
+
+ GenericGroup(String name, String displayName, Limits limits) {
+ super(name, displayName, limits);
+ }
+
+ @Override
+ protected TezCounter newCounter(String name, String displayName, long value) {
+ return new GenericCounter(name, displayName, value);
+ }
+
+ @Override
+ protected TezCounter newCounter() {
+ return new GenericCounter();
+ }
+
+ @Override
+ public CounterGroupBase<TezCounter> getUnderlyingGroup() {
+ return this;
+ }
+ }
+
+ // Mix file system group implementation into the CounterGroup interface
+ private static class FileSystemGroup extends FileSystemCounterGroup<TezCounter>
+ implements CounterGroup {
+
+ @Override
+ protected TezCounter newCounter(String scheme, FileSystemCounter key) {
+ return new FSCounter(scheme, key);
+ }
+
+ @Override
+ public CounterGroupBase<TezCounter> getUnderlyingGroup() {
+ return this;
+ }
+ }
+
+ /**
+ * Provide factory methods for counter group factory implementation.
+ * See also the GroupFactory in
+ * {@link org.apache.hadoop.TezCounters.Counters mapred.Counters}
+ */
+ private static class GroupFactory
+ extends CounterGroupFactory<TezCounter, CounterGroup> {
+
+ @Override
+ protected <T extends Enum<T>>
+ FrameworkGroupFactory<CounterGroup>
+ newFrameworkGroupFactory(final Class<T> cls) {
+ return new FrameworkGroupFactory<CounterGroup>() {
+ @Override public CounterGroup newGroup(String name) {
+ return new FrameworkGroupImpl<T>(cls); // impl in this package
+ }
+ };
+ }
+
+ @Override
+ protected CounterGroup newGenericGroup(String name, String displayName,
+ Limits limits) {
+ return new GenericGroup(name, displayName, limits);
+ }
+
+ @Override
+ protected CounterGroup newFileSystemGroup() {
+ return new FileSystemGroup();
+ }
+ }
+
+ private static final GroupFactory groupFactory = new GroupFactory();
+
+ /**
+ * Default constructor
+ */
+ public TezCounters() {
+ super(groupFactory);
+ }
+
+ /**
+ * Construct the Counters object from the another counters object
+ * @param <C> the type of counter
+ * @param <G> the type of counter group
+ * @param counters the old counters object
+ */
+ public <C extends TezCounter, G extends CounterGroupBase<C>>
+ TezCounters(AbstractCounters<C, G> counters) {
+ super(counters, groupFactory);
+ }
+}