Use Case: How to Implement Hive Hooks to Optimize a Data Lake

Data About Data

The important difference between data lakes and data swamps is prudently organized data leads to an efficient lake while a swamp is just data that is either over-replicated or siloed by its users.  Getting the information on how the production data is being used across organization can not only be beneficial in building a well-organized data lake but it will also help data engineers to fine-tune the data pipelines or data itself.

To understand how data is consumed, we need to figure out answers to some basic questions like:

• Which datasets (tables/views/DBs) are accessed frequently?
• When are the queries run most frequently?
• Which users or applications are heavily utilizing the resources?
• What type of queries are running frequently?

The most accessed object can easily benefit from optimization like compression, columnar file format, or data decomposition. A separate queue can be assigned to heavy-resource-utilizing apps or users to balance the load on a cluster. Cluster resources can be scaled up during the timeframe when a large number of queries are mostly run to meet SLAs and scaled down during low usage tide to save cost.

Hive Hooks are convenient ways to answer some of the above questions and more!

Hooks

Hooks are a mechanism that allows you to modify the behavior of a program. It’s a technique to intercept functional calls, messages, or events in an application. Hive provides many different types of hooks, a few of them are listed below:

• Pre- and Post-Execution.
• Pre- and Post-Drive Run.
• Execution-failure.
• Pre- and Post-Semantic Analysis.
• Post-Execution ORC file dump.
• QueryLifeTime.
• Redactor

Each type of hook can be invoked at a specific event and can be customized to take different actions based on a use case. For example, a pre-execution hook is called before executing the physical query plan and a redactor hook is invoked before submitting a query to job.xml to redact sensitive information. Apache Atlas has one of the most popular implementations of Hive hooks which listens to create/update/delete operations in Hive and updates the metadata in Atlas through Kafka notifications.

Implementation 

Pre-execution hooks can be created by implementing the ExecuteWithHookContext interface. This is an empty interface which simply calls the run method along with HookContext. HookContext has a lot of information about the query, hive instance, and user. This information can be easily leveraged to detect how a data lake is being utilized by its users.

public HookContext(QueryPlan queryPlan, QueryState queryState,
      Map<String, ContentSummary> inputPathToContentSummary, String userName, String ipAddress,
      String hiveInstanceAddress, String operationId, String sessionId, String threadId,
      boolean isHiveServerQuery, PerfLogger perfLogger, QueryInfo queryInfo) throws Exception {

Any hook implementing HookContext gets a query plan (QueryPlan). Looking under the hood of a QueryPlan, there are many getters implemented to gather information about the query. To name a few:

•  getQueryProperties – Gets detailed information about a query including whether the query has joins, group-bys, analytical functions, or any sorting/ordering operations.
•  getQueryStartTime – Returns the start time for a query.
•  getOperationName – Returns the type of operation being performed by a query for  example, CREATETABLE, DROPTABLE, ALTERDATABASE, etc.,
•  getQueryStr – Returns the query as a string.

To create our own Hive hook, we just need a class that implements ExecuteWithHookContext and overrides its run method with our custom logic to capture data.

public class CustomHook implements ExecuteWithHookContext {
private static final Logger logger = Logger.getLogger(CustomHook.class.getName());

public void run(HookContext hookContext) throws Exception {

assert (hookContext.getHookType() == HookType.PRE_EXEC_HOOK);

SessionState ss = SessionState.get();
UserGroupInformation ugi = hookContext.getUgi();
Set<ReadEntity> inputs = hookContext.getInputs();
QueryPlan plan = hookContext.getQueryPlan();
this.run(ss, ugi, plan, inputs);
}

SessionState  and UserGroupInformation are required to gather information about a Hive session and its users.

public void run(SessionState sess, UserGroupInformation ugi, QueryPlan qpln, Set<ReadEntity> inputs)
throws Exception {
if (sess != null) {
String qid = sess.getQueryId() == null ? qpln.getQueryId() : sess.getQueryId();
String QueryID = qid;
String Query = sess.getCmd().trim();
String QueryType = sess.getCommandType();
//  get all information about query 
if (qpln != null) {
Long Query_Start_Time = qpln.getQueryStartTime();
QueryProperties queryProps = qpln.getQueryProperties();
if (queryProps != null) {
boolean Has_Join = queryProps.hasJoin();
boolean Has_Group_By = queryProps.hasGroupBy();
boolean Has_Sort_By = queryProps.hasSortBy();
boolean Has_Order_By = queryProps.hasOrderBy();
boolean Has_Distribute_By = queryProps.hasDistributeBy();
boolean Has_Cluster_By = queryProps.hasClusterBy();
boolean Has_Windowing = queryProps.hasWindowing();
}
}
// get user id 
String username = sess.getUserName() == null ? ugi.getUserName() : sess.getUserName();
// get list of database@table names
List<String> tables = new ArrayList<String>();
for (Object o : inputs) {
tables.add(o.toString());
}

// Add logic here to format logging msg
        // logger.info(msg)
}
}

The compiled jar should be added to the Hive classpath before a hook can be assigned. A jar can be added in a location defined by the hive-site.xml property hive.aux.jars.path. A pre-execution hook can be set using the property hive.exec.pre.hooks to the class of a custom hook. Using the Hive CLI, we can do the following: 

set hive.exec.pre.hooks=com.myApp.CustomHook;  

Once this pre-execution hook is set, the code for CustomHook should execute for each query for each user. Information gathered by CustomHook can be logged as comma separated values in a common logging repository and pulled out later in any BI tool or Excel file to figure out various statistics about data lake usage patterns.  

Caveats

• Although hooks are a great way to capture information, they can add latency to query executions. Processing in a hook can be kept to a minimum to avoid this overhead.
• Information about processing done on Hive tables through Spark’s HiveContext cannot be captured through Hive hooks. Spark provides its own hook mechanism.