diff --git a/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveHookEventProtoPartialBuilder.java b/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveHookEventProtoPartialBuilder.java new file mode 100644 index 00000000000..350328d2da7 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveHookEventProtoPartialBuilder.java @@ -0,0 +1,55 @@ +/* + * + * * 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.hadoop.hive.ql.hooks; + +import java.util.Map; + +import org.apache.hadoop.hive.ql.hooks.HiveProtoLoggingHook.OtherInfoType; +import org.apache.hadoop.hive.ql.hooks.proto.HiveHookEvents; +import org.json.JSONObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class HiveHookEventProtoPartialBuilder { + private static final Logger LOG = LoggerFactory.getLogger(HiveHookEventProtoPartialBuilder.class.getName()); + private final HiveHookEvents.HiveHookEventProto event; + private final Map otherInfo; + + public HiveHookEventProtoPartialBuilder(HiveHookEvents.HiveHookEventProto.Builder builder, Map otherInfo) { + this.event = builder.buildPartial(); + this.otherInfo = otherInfo; + } + + public HiveHookEvents.HiveHookEventProto build() { + HiveHookEvents.HiveHookEventProto.Builder builder = HiveHookEvents.HiveHookEventProto.newBuilder(); + for (Map.Entry each : otherInfo.entrySet()) { + OtherInfoType type = each.getKey(); + JSONObject json = each.getValue(); + try { + // json conversion can be expensive, doing it separately + HiveProtoLoggingHook.EventLogger.addMapEntry(builder, type, json.toString()); + } catch (Exception e) { + LOG.error("Unexpected exception while serializing json.", e); + } + } + return builder.mergeFrom(event).build(); + } +} diff --git a/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveProtoLoggingHook.java b/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveProtoLoggingHook.java index 86a68008515..97e35b5d000 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveProtoLoggingHook.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/hooks/HiveProtoLoggingHook.java @@ -88,6 +88,7 @@ import java.time.LocalDate; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; @@ -252,7 +253,7 @@ void handle(HookContext hookContext) { LOG.debug("Not logging events of operation type : {}", plan.getOperationName()); return; } - HiveHookEventProto event; + HiveHookEventProtoPartialBuilder event; switch (hookContext.getHookType()) { case PRE_EXEC_HOOK: event = getPreHookEvent(hookContext); @@ -309,7 +310,8 @@ private boolean maybeRolloverWriterForDay() throws IOException { } private static final int MAX_RETRIES = 2; - private void writeEvent(HiveHookEventProto event) { + private void writeEvent(HiveHookEventProtoPartialBuilder builder) { + HiveHookEventProto event = builder.build(); for (int retryCount = 0; retryCount <= MAX_RETRIES; ++retryCount) { try { if (eventPerFile) { @@ -349,7 +351,7 @@ private void writeEvent(HiveHookEventProto event) { } } - private HiveHookEventProto getPreHookEvent(HookContext hookContext) { + private HiveHookEventProtoPartialBuilder getPreHookEvent(HookContext hookContext) { QueryPlan plan = hookContext.getQueryPlan(); LOG.info("Received pre-hook notification for: " + plan.getQueryId()); @@ -359,6 +361,7 @@ private HiveHookEventProto getPreHookEvent(HookContext hookContext) { List tezTasks = Utilities.getTezTasks(plan.getRootTasks()); ExecutionMode executionMode = getExecutionMode(plan, mrTasks, tezTasks); + Map otherInfo = new HashMap<>(); HiveHookEventProto.Builder builder = HiveHookEventProto.newBuilder(); builder.setEventType(EventType.QUERY_SUBMITTED.name()); builder.setTimestamp(plan.getQueryStartTime()); @@ -380,7 +383,7 @@ private HiveHookEventProto getPreHookEvent(HookContext hookContext) { JSONObject queryObj = new JSONObject(); queryObj.put("queryText", plan.getQueryStr()); queryObj.put("queryPlan", getExplainPlan(plan, conf, hookContext)); - addMapEntry(builder, OtherInfoType.QUERY, queryObj.toString()); + otherInfo.put(OtherInfoType.QUERY, queryObj); } catch (Exception e) { LOG.error("Unexpected exception while serializing json.", e); } @@ -417,14 +420,15 @@ private HiveHookEventProto getPreHookEvent(HookContext hookContext) { for (Map.Entry setting : conf) { confObj.put(setting.getKey(), setting.getValue()); } - addMapEntry(builder, OtherInfoType.CONF, confObj.toString()); - return builder.build(); + otherInfo.put(OtherInfoType.CONF, confObj); + return new HiveHookEventProtoPartialBuilder(builder, otherInfo); } - private HiveHookEventProto getPostHookEvent(HookContext hookContext, boolean success) { + private HiveHookEventProtoPartialBuilder getPostHookEvent(HookContext hookContext, boolean success) { QueryPlan plan = hookContext.getQueryPlan(); LOG.info("Received post-hook notification for: " + plan.getQueryId()); + Map other = new HashMap<>(); HiveHookEventProto.Builder builder = HiveHookEventProto.newBuilder(); builder.setEventType(EventType.QUERY_COMPLETED.name()); builder.setTimestamp(clock.getTime()); @@ -439,12 +443,11 @@ private HiveHookEventProto getPostHookEvent(HookContext hookContext, boolean suc for (String key : hookContext.getPerfLogger().getEndTimes().keySet()) { perfObj.put(key, hookContext.getPerfLogger().getDuration(key)); } - addMapEntry(builder, OtherInfoType.PERF, perfObj.toString()); - - return builder.build(); + other.put(OtherInfoType.PERF, perfObj); + return new HiveHookEventProtoPartialBuilder(builder, other); } - private void addMapEntry(HiveHookEventProto.Builder builder, OtherInfoType key, String value) { + public static void addMapEntry(HiveHookEventProto.Builder builder, OtherInfoType key, String value) { if (value != null) { builder.addOtherInfo( MapFieldEntry.newBuilder().setKey(key.name()).setValue(value).build()); diff --git a/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveHookEventProtoPartialBuilder.java b/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveHookEventProtoPartialBuilder.java new file mode 100644 index 00000000000..1bc0e2afcbd --- /dev/null +++ b/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveHookEventProtoPartialBuilder.java @@ -0,0 +1,81 @@ +/* + * + * * 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.hadoop.hive.ql.hooks; + +import static java.util.Collections.singletonList; +import static org.junit.Assert.*; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.hadoop.hive.ql.hooks.proto.HiveHookEvents; +import org.json.JSONObject; +import org.junit.Test; + +public class TestHiveHookEventProtoPartialBuilder { + private static final String QUERY_1 = "query1"; + private static final String HIVE = "hive"; + private static final String LLAP = "llap"; + private static final String TEZ = "tez"; + private static final long TIMESTAMP = System.currentTimeMillis(); + + @Test + public void testEquality() { + JSONObject json = new JSONObject(); + json.put("key1", "value1"); + json.put("key2", "value2"); + json.put("key3", "value3"); + HiveHookEvents.HiveHookEventProto event1 = buildWithOtherInfo(json); + HiveHookEvents.HiveHookEventProto event2 = buildIn2Steps(json); + assertArrayEquals(event1.toByteArray(), event2.toByteArray()); + } + + private HiveHookEvents.HiveHookEventProto buildWithOtherInfo(JSONObject json) { + return HiveHookEvents.HiveHookEventProto + .newBuilder() + .setEventType(HiveProtoLoggingHook.EventType.QUERY_SUBMITTED.name()) + .setTimestamp(TIMESTAMP) + .setHiveQueryId(QUERY_1) + .setUser(HIVE) + .setRequestUser(HIVE) + .setQueue(LLAP) + .setExecutionMode(TEZ) + .addAllOtherInfo(singletonList(HiveHookEvents.MapFieldEntry.newBuilder() + .setKey(HiveProtoLoggingHook.OtherInfoType.CONF.name()) + .setValue(json.toString()).build())) + .build(); + } + + private HiveHookEvents.HiveHookEventProto buildIn2Steps(JSONObject json) { + HiveHookEvents.HiveHookEventProto.Builder builder = HiveHookEvents.HiveHookEventProto + .newBuilder() + .setEventType(HiveProtoLoggingHook.EventType.QUERY_SUBMITTED.name()) + .setTimestamp(TIMESTAMP) + .setHiveQueryId(QUERY_1) + .setUser(HIVE) + .setRequestUser(HIVE) + .setQueue(LLAP) + .setExecutionMode(TEZ); + Map otherInfo = new HashMap<>(); + otherInfo.put(HiveProtoLoggingHook.OtherInfoType.CONF, json); + return new HiveHookEventProtoPartialBuilder(builder, otherInfo).build(); + } +} \ No newline at end of file diff --git a/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveProtoLoggingHook.java b/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveProtoLoggingHook.java index add4b6863d8..d0b67948648 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveProtoLoggingHook.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/hooks/TestHiveProtoLoggingHook.java @@ -21,6 +21,7 @@ import java.io.EOFException; import java.io.IOException; import java.util.ArrayList; +import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -46,6 +47,7 @@ import org.apache.hadoop.hive.ql.plan.HiveOperation; import org.apache.hadoop.hive.ql.plan.MapWork; import org.apache.hadoop.hive.ql.plan.TezWork; +import org.apache.hadoop.hive.ql.plan.api.Query; import org.apache.hadoop.mapreduce.MRJobConfig; import org.apache.hadoop.yarn.util.SystemClock; import org.apache.tez.dag.api.TezConfiguration; @@ -84,6 +86,7 @@ public void setup() throws Exception { QueryPlan queryPlan = new QueryPlan(HiveOperation.QUERY) {}; queryPlan.setQueryId("test_queryId"); queryPlan.setQueryStartTime(1234L); + queryPlan.setQueryString("SELECT * FROM t WHERE i > 10"); queryPlan.setRootTasks(new ArrayList<>()); queryPlan.setInputs(new HashSet<>()); queryPlan.setOutputs(new HashSet<>());