Skip to content

HIVE-23375: Introduce TezCounter to track MJ HashTable Load time #1005

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pgaref wants to merge 1 commit into from
Closed

HIVE-23375: Introduce TezCounter to track MJ HashTable Load time #1005

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
*/
public interface HashTableLoader {

enum HashTableLoaderCounters {
HASHTABLE_LOAD_TIME_MS
};

void init(ExecMapperContext context, MapredContext mrContext, Configuration hconf,
MapJoinOperator joinOp);

Expand Down
11 changes: 11 additions & 0 deletions ql/src/java/org/apache/hadoop/hive/ql/exec/tez/HashTableLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,10 @@

import org.apache.hadoop.hive.llap.LlapDaemonInfo;
import org.apache.hadoop.hive.ql.exec.MemoryMonitorInfo;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.mapjoin.MapJoinMemoryExhaustionError;
import org.apache.tez.common.counters.TezCounter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -66,6 +69,7 @@ public class HashTableLoader implements org.apache.hadoop.hive.ql.exec.HashTable
private MapJoinDesc desc;
private TezContext tezContext;
private String cacheKey;
private TezCounter htLoadCounter;

@Override
public void init(ExecMapperContext context, MapredContext mrContext, Configuration hconf,
Expand All @@ -74,6 +78,10 @@ public void init(ExecMapperContext context, MapredContext mrContext, Configurati
this.hconf = hconf;
this.desc = joinOp.getConf();
this.cacheKey = joinOp.getCacheKey();
String counterGroup = HiveConf.getVar(hconf, HiveConf.ConfVars.HIVECOUNTERGROUP);
String vertexName = hconf.get(Operator.CONTEXT_NAME_KEY, "");
String counterName = Utilities.getVertexCounterName(HashTableLoaderCounters.HASHTABLE_LOAD_TIME_MS.name(), vertexName);
this.htLoadCounter = tezContext.getTezProcessorContext().getCounters().findCounter(counterGroup, counterName);
}

@Override
Expand Down Expand Up @@ -238,6 +246,7 @@ public void load(MapJoinTableContainer[] mapJoinTables,
cacheKey, tableContainer.getClass().getSimpleName(), pos);

tableContainer.setSerde(keyCtx, valCtx);
long startTime = System.currentTimeMillis();
while (kvReader.next()) {
tableContainer.putRow((Writable) kvReader.getCurrentKey(), (Writable) kvReader.getCurrentValue());
numEntries++;
Expand All @@ -258,6 +267,8 @@ public void load(MapJoinTableContainer[] mapJoinTables,
}
}
}
long delta = System.currentTimeMillis() - startTime;
htLoadCounter.increment(delta);
tableContainer.seal();
mapJoinTables[pos] = tableContainer;
if (doMemCheck) {
Expand Down
11 changes: 11 additions & 0 deletions .../org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/VectorMapJoinFastHashTableLoader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,10 @@

import org.apache.hadoop.hive.llap.LlapDaemonInfo;
import org.apache.hadoop.hive.ql.exec.MemoryMonitorInfo;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.mapjoin.MapJoinMemoryExhaustionError;
import org.apache.tez.common.counters.TezCounter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -54,6 +57,7 @@ public class VectorMapJoinFastHashTableLoader implements org.apache.hadoop.hive.
protected MapJoinDesc desc;
private TezContext tezContext;
private String cacheKey;
private TezCounter htLoadCounter;

@Override
public void init(ExecMapperContext context, MapredContext mrContext,
Expand All @@ -62,6 +66,10 @@ public void init(ExecMapperContext context, MapredContext mrContext,
this.hconf = hconf;
this.desc = joinOp.getConf();
this.cacheKey = joinOp.getCacheKey();
String counterGroup = HiveConf.getVar(hconf, HiveConf.ConfVars.HIVECOUNTERGROUP);
String vertexName = hconf.get(Operator.CONTEXT_NAME_KEY, "");
String counterName = Utilities.getVertexCounterName(HashTableLoaderCounters.HASHTABLE_LOAD_TIME_MS.name(), vertexName);
this.htLoadCounter = tezContext.getTezProcessorContext().getCounters().findCounter(counterGroup, counterName);
}

@Override
Expand Down Expand Up @@ -126,6 +134,7 @@ public void load(MapJoinTableContainer[] mapJoinTables,
cacheKey, vectorMapJoinFastTableContainer.getClass().getSimpleName(), pos);

vectorMapJoinFastTableContainer.setSerde(null, null); // No SerDes here.
long startTime = System.currentTimeMillis();
while (kvReader.next()) {
vectorMapJoinFastTableContainer.putRow((BytesWritable)kvReader.getCurrentKey(),
(BytesWritable)kvReader.getCurrentValue());
Expand All @@ -147,6 +156,8 @@ public void load(MapJoinTableContainer[] mapJoinTables,
}
}
}
long delta = System.currentTimeMillis() - startTime;
htLoadCounter.increment(delta);

vectorMapJoinFastTableContainer.seal();
mapJoinTables[pos] = vectorMapJoinFastTableContainer;
Expand Down
6 changes: 5 additions & 1 deletion ql/src/java/org/apache/hadoop/hive/ql/hooks/PostExecTezSummaryPrinter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,11 @@ public void run(HookContext hookContext) throws Exception {
if (hiveCountersGroup.equals(group.getDisplayName())) {
console.printInfo(tezTask.getId() + " HIVE COUNTERS:", false);
for (TezCounter counter : group) {
console.printInfo(" " + counter.getDisplayName() + ": " + counter.getValue(), false);
// HIVE Counter names are picked at runtime so cannot rely on testSafeCounterNames like in LlapIOCounters
// Here we just filter out time counters (like HASHTABLE_LOAD_TIME_MS) that may differ across runs
if (!counter.getName().contains("TIME")) {
console.printInfo(" " + counter.getDisplayName() + ": " + counter.getValue(), false);
}
}
} else if (group.getName().equals(HiveInputCounters.class.getName())) {
console.printInfo(tezTask.getId() + " INPUT COUNTERS:", false);
Expand Down