How does Map-side aggregation work in Hive?

What is map-side aggregation in hive?

Hive trys to optimize the query execution by performing a map-side aggregations when certain conditions are met in the query execution.

What exactly is the optimization being perfromed?

The optimization does a partial aggregation inside of the mapper, which results in the mapper output fewer rows, reducing the data that Hadoop needs to sort and distribute to the reducers.

How does the optimization/map-side aggregation work?

Mappers store each map output key with the corresponding partial aggregation,  Periodically, the mappers will output the pairs (#{token}, #{token_count}). The Hadoop framework again sorts these pairs and the reducers aggregates these values. 

So what is the tradeoff of using map-side aggregations?

There is the need to keep a map of all tokens in memory. This leads in increase in the Main Memory needed to execute the Mapper

How does hive decide to apply map-side aggregation optimizations?

By default, Hive will try to use the map-side aggregation optimization but it falls back vanilla MR approach if the hash map is not producing enough of a memory savings. 

Decision policy:

After processing the first 100,000 (hive.groupby.mapaggr.checkinterval) records in the mapper, hive checks if the size of the hash map exceeds 50%(hive.map.aggr.hash.min.reduction) of the number of records, If true it aborts Map side aggregations.  The respective configurations can be modified to control the map-side aggregation behaviour.

 So what is hive.map.aggr.hash.percentmemory  for ?

We know the trade-off with Map-side aggregations is the hash Map need to be stored in memory. As a preventive measure to avoid "Out of Memory Exception " in the mapper by 
the size explosion in the aggregation hash map, hive.map.aggr.hash.percentmemory
can be used to control the flush of the hash map to the reducer wherever hash map size exceeds the percentage specified in the parameter compared to the total memory available to the mapper.

This, however, is an estimate based on the number of rows and the expected size of each row, so if the memory usage is per row is unexpectedly high, the mappers may run out of memory before the hash map is flushed to the reducers.

Orchestrating Hadoop Workflows - What your workflow tool should look like.

This post illustrates what are the most important qualities to look for while deciding a Big data workflow tool.

1. Schedule the ETL process to ingest incremental data from various streams.
   
   
   Ability to schedule is the most elementary requirement of any orchestration tool, and the tool
   must be abel to ingest and process data simultaneously form various sources to various endpoints. It should support most common sources and endpoints out of the box.
   
   
2. Notifications on failure of ETL process.
   
   
   Notifications and Alerts are a must, a failure can setback you to backfill and can drastically increase the infrastructure costs (especially if u pay by the hour for the cluster). So a good reliable notification mechanism is essential and a must have.
   
   
3. Optimization on resources usage for daily ingestion.
   
   
   Not directly supported by the workflow tool, but the tool should persist the execution time and trends for a task, which can be used to optimize your tasks. This data becomes important when you have run the workflow sufficiently and can understand the time trends.
   
   
4. Retry on failures, make ur tasks as Idempotent incremental loads
   
   
   The ownership of making the workflow idempotent is on the developer and not the tool. The tool should support retires.
   
   
5. Zero/Minimal Manual intervention
   
   
6. Self recovering/ rescheduling system
   
   
7. Pause the workflow
   
   
   The tool should support you , when needed to pause the execution after the current running task. This enables you to fix any issue with the next task in your workflow without having to fail the workflow.
   
   
8. Automatic Data audits, for incremental loaded data.
   
   
   Should support audits on tasks and help make decision in the workflow.
   
   
   
   

Submit Hive queries to different Queues

Hive Execution Engine is MapReduce 

set mapreduce.job.queuename=<queuename>; 

Hive Execution Engine is Tez

set tez.queue.name=<queuename>; 

Hive Default Field Delimiter

\U0001  ( CTRL+SHIFT+A in windows) 

Hive RuntimeException

"Failed with exception java.io.IOException:java.lang.RuntimeException: serious problem"

If the hive cluster is a Kerberos secured cluster, disable Hive optimizations.

set hive.fetch.task.conversion=none;

Bucket Map Join in Hive

Shuffle Join and Map Joins are most common joins used in Hive, these joins might just not be the best solution to all the cases.

understanding Bucket joins can be handy to improve join execution speeds based on the join columns.

Bucket Map Joins as the name indicates is a Mapper only joins. (not to be confused with map joins/broadcast joins).

So when do we use Bucket Map join?

1. We have Large tables as part of our n-way join.
2. The tables are bucketed on join columns.
3. Data is unsorted
4. The number of buckets in the tables are multiples of eachother. 

So , Large unsorted tables which have been bucketed on join columns are good candidates. Other way is to design the table to be bucketed based on the use case. 

Bucket map joins are not the default behavior of hive, hence we need to suggest hive by setting the bucketmapjoin to true.

set hive.optimize.bucketmapjoin = true; Select T1.key, T1.value FROM T1 JOIN T2 ON T1.key = T2.key;

We need to bucket Map join on table T1 and T2. The tables must be bucketed on the column "Key" and the number of buckets in both the tables should be multiples of each other.

Bucket join now has an advantage of choosing only the required Buckets from T2 instead of the entire T2 table.

The Mapper will be able to access only the required buckets from the n-1 tables in the n-way join. This results to join happening in the Mapper, the expensive shuffle part is skipped.