diff --git common/src/java/org/apache/hadoop/hive/conf/HiveConf.java common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index 7a8517b..07f2a96 100644 --- common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -1196,6 +1196,9 @@ public void setSparkConfigUpdated(boolean isSparkConfigUpdated) { HIVETEZLOGLEVEL("hive.tez.log.level", "INFO", "The log level to use for tasks executing as part of the DAG.\n" + "Used only if hive.tez.java.opts is used to configure Java options."), + HIVEQUERYNAME ("hive.query.name", null, + "This named is used by Tez to set the dag name. This name in turn will appear on \n" + + "the Tez UI representing the work that was done."), HIVEENFORCEBUCKETING("hive.enforce.bucketing", false, "Whether bucketing is enforced. If true, while inserting into the table, bucketing is enforced."), diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java index 6dcfe8d..80fbc92 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java @@ -50,6 +50,7 @@ import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.ql.Context; import org.apache.hadoop.hive.ql.ErrorMsg; +import org.apache.hadoop.hive.ql.QueryPlan; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.Utilities; import org.apache.hadoop.hive.ql.exec.mr.ExecMapper; @@ -1216,6 +1217,21 @@ private void setupAutoReducerParallelism(TezEdgeProperty edgeProp, Vertex v) } } + public String createDagName(Configuration conf, QueryPlan plan) { + String name = HiveConf.getVar(conf, HiveConf.ConfVars.HIVEQUERYNAME); + + if (name == null) { + name = conf.get("mapred.job.name"); + } + + if (name == null) { + name = plan.getQueryId(); + } + + assert name != null; + return name; + } + private DagUtils() { // don't instantiate } diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java index a6d911d..270eede 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java @@ -167,9 +167,8 @@ public int execute(DriverContext driverContext) { if (driverContext.getCtx() == null) { boolean a = false; } - CallerContext callerContext = CallerContext.create("HIVE", - conf.getLogIdVar(SessionState.get().getSessionId()) + " " - + conf.getVar(HiveConf.ConfVars.HIVEQUERYID), + CallerContext callerContext = CallerContext.create( + "HIVE", queryPlan.getQueryId(), "HIVE_QUERY_ID", queryPlan.getQueryStr()); dag.setCallerContext(callerContext); @@ -311,7 +310,10 @@ DAG build(JobConf conf, TezWork work, Path scratchDir, FileSystem fs = scratchDir.getFileSystem(conf); // the name of the dag is what is displayed in the AM/Job UI - DAG dag = DAG.create(work.getName()); + String dagName = utils.createDagName(conf, queryPlan); + + LOG.info("Dag name: " + dagName); + DAG dag = DAG.create(dagName); // set some info for the query JSONObject json = new JSONObject(new LinkedHashMap()).put("context", "Hive") diff --git ql/src/java/org/apache/hadoop/hive/ql/hooks/ATSHook.java ql/src/java/org/apache/hadoop/hive/ql/hooks/ATSHook.java index 38b6b5d..578df33 100644 --- ql/src/java/org/apache/hadoop/hive/ql/hooks/ATSHook.java +++ ql/src/java/org/apache/hadoop/hive/ql/hooks/ATSHook.java @@ -50,13 +50,14 @@ private static final Logger LOG = LoggerFactory.getLogger(ATSHook.class.getName()); private static final Object LOCK = new Object(); + private static final int VERSION = 2; private static ExecutorService executor; private static TimelineClient timelineClient; private enum EntityTypes { HIVE_QUERY_ID }; private enum EventTypes { QUERY_SUBMITTED, QUERY_COMPLETED }; private enum OtherInfoTypes { - QUERY, STATUS, TEZ, MAPRED, INVOKER_INFO, THREAD_NAME + QUERY, STATUS, TEZ, MAPRED, INVOKER_INFO, THREAD_NAME, VERSION }; private enum PrimaryFilterTypes { user, requestuser, operationid }; private static final int WAIT_TIME = 3; @@ -193,6 +194,7 @@ TimelineEntity createPreHookEvent(String queryId, String query, JSONObject expla atsEntity.addOtherInfo(OtherInfoTypes.MAPRED.name(), numMrJobs > 0); atsEntity.addOtherInfo(OtherInfoTypes.INVOKER_INFO.name(), logID); atsEntity.addOtherInfo(OtherInfoTypes.THREAD_NAME.name(), Thread.currentThread().getName()); + atsEntity.addOtherInfo(OtherInfoTypes.VERSION.name(), VERSION); return atsEntity; }