You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 2 Next »

Abstract

Apache Hadoop is a framework for the distributed processing of large data sets using clusters of computers typically composed of commodity hardware. Over last few years Apache Hadoop has become the de facto platform for distributed data processing using commodity hardware. Apache Hive is a popular SQL interface for data processing using Apache Hadoop.

 

User submitted SQL query is converted by Hive to physical operator tree which is optimized and converted to Tez Jobs and is then executed on Hadoop cluster. Distributed SQL query processing in Hadoop differs from conventional relational query engine when it comes to handling of intermediate result sets. Hive query processing often requires sorting and reassembling of intermediate result set; this is called shuffling in Hadoop parlance.

 

Most of the existing query optimizations in Hive are about minimizing shuffling cost. Currently user would have to submit an optimized query to Hive with right join order for query to be executed efficiently. Logical optimizations in Hive are limited to filter push down, projection pruning and partition pruning. Cost based logical optimizations can significantly improve Apache Hive’s query latency and ease of use.

 

Join reordering and join algorithm selection are few of the optimizations that can benefit from a cost based optimizer. Cost based optimizer would free up user from having to rearrange joins in the right order or from having to specify join algorithm by using query hints and configuration options. This can potentially free up users to model their reporting and ETL needs close to business process without having to worry about query optimizations.

 

Optiq is an open source cost based query optimizer and query execution framework. Optiq currently has more than fifty query optimization rules that can rewrite query tree, and an efficient plan pruner that can select cheapest query plan in an optimal manner. In this paper we discuss how Optiq can be used to introduce Cost Based Logical Optimizer (CBO) in to Apache Hive.

 

CBO will be introduced in to Hive in a Phased manner. In the first phase, Optiq would be used to reorder joins and to pick right join algorithm so as to reduce query latency. Table cardinality and Boundary statistics will be used for this cost based optimizations.

1. INTRODUCTION

Hive is a data-warehousing infrastructure on top of Apache Hadoop. Hive takes advantage of Hadoop’s massive scale out and fault tolerance capabilities for data storage and processing on commodity hardware. Hive is designed to enable easy data summarization, ad-hoc querying and analysis of large volumes of data. Hive SQL is the declarative query language, which enables users familiar with SQL to do ad-hoc querying, summarization and data analysis easily.

In past Hadoop jobs tended to have high latencies and incurred substantial overheads in job submission and scheduling. As a result - latency for Hive queries was generally very high even when data sets involved were very small. As a result Hive was typically used for ETL and not much for interactive queries. With Hadoop2 and Tez the overheads for job submission and job scheduling have gone down significantly. In Hadoop version one, the jobs that could be executed could only be Map-Reduce Jobs.  With Hadoop2 and Tez that limitation no longer apply.

In Hadoop the output of mapper is sorted and sometimes persisted on the local disk of the mapper. This sorted mapper output is then send to appropriate reducer which then combines sorted results from different mappers.  While executing multiple map-reduce jobs where output of one job needs to be consumed by the successive map-reduce job, the output of preceding map-reduce job needs to be persisted into HDFS; this persistence is costly as the data needs to be copied to other nodes based on the replication factor of HDFS.

Hive on top of Hadoop version 1 often had to submit multiple map-reduce jobs to complete query processing. This Map-Reduce job pipeline degraded performance, as the intermediate result set now needs to be persisted to fault tolerant HDFS. Also submitting jobs and scheduling them were relatively costly operations. With Hadoop2 and Tez the cost of job submission and scheduling is minimized. Also Tez does not restrict the job to be only Map followed by Reduce; this implies all of the query execution can be done in a single job without having to cross job boundaries. This would result in a significant cost savings, as the intermediate result sets need not be persisted to HDFS or to even local disk.

Query optimizations in a relational query engine can be broadly classified as logical query optimizations and physical query optimizations. Logical query optimizations generally refer to query optimizations that can be derived based on relational algebra independent of the physical layer in which query is executed. Physical query optimizations are query optimizations that are cognizant of physical layer primitives. For Hive, the physical layer implies Map-Reduce and Tez primitives.

Currently logical query optimizations in Hive can be broadly categorized as follows:

  • Projection Pruning
  • Deducing Transitive Predicates
  • Predicate Push down
  • Merging of Select-Select, Filter-Filter in to single operator
  • Multi-way Join
  • Query Rewrite to accommodate for Join skew on some column values

 

Physical optimizations in Hive can be broadly classified as follows:

  • Partition Pruning
  • Scan pruning based on partitions and bucketing
  • Scan pruning if query is based on sampling
  • Apply Group By on the map side in some cases
  • Perform Join on the Mapper
  • Optimize Union so that union can be performed on map side only
  • Decide which table to stream last, based on user hint, in a multi way join
  • Remove unnecessary reduce sink operators
  • For queries with limit clause, reduce the number of files that needs to be scanned for the table.
  • For queries with Limit clause, restrict the output coming from mapper by restricting what Reduce Sink operator generates.
  • Reduce the number of Tez jobs needed for answering user submitted SQL query
  • Avoid Map-Reduce jobs in case of simple fetch query
  • For simple fetch queries with aggregation, perform aggregation without Map-Reduce tasks
  • Rewrite Group By Query to use index table instead of original table
  • Use Index scans when predicates above table scan is equality predicates and columns in predicate have indexes on it.

In Hive most of the optimizations are not based on the cost of query execution. Most of the optimizations do not rearrange the operator tree except for filter push down and operator merging.  Most of the operator tree mutation is for removing reduce-sink and reducer operator.  Listed below are some of optimization decisions that can benefit from a CBO:

  • How to order Join
  • What algorithm to use for a given Join
  • Should the intermediate result be persisted or should it be recomputed on operator failure.
  • The degree of parallelism at any operator (specifically number of reducers to use).
  • Semi Join selection

Optiq is an open source, Apache Licensed, query planning and execution framework. Many pieces of Optiq are derived from Eigenbase Project. Optiq has optional JDBC server, query parser and validator, query optimizer and pluggable data source adapters. One of the available Optiq optimizer is a cost based optimizer based on volcano paper. Currently different pieces of Optiq is used in following projects/products:

  • Apache Drill
  • Cascading (Lingual)
  • Lucid DB
  • Mondrian/Pentaho

 

Optiq currently has over fifty cost based optimization rules. Some of the prominent cost based optimization rules are listed below:

  • Push Join through Union
  • Push Filter past Table Function
  • Join Reordering
  • Semi Join selection
  • Push Aggregate through Union
  • Pull Aggregate through Union
  • Pull Constants through Aggregate
  • Merge Unions

In this document we propose to use Optiq’s cost based optimizer, Volcano, to perform Cost Based Optimizations in Hive. We propose to implement Optiq based CBO in a phased manner. Note here that proposal is to use Optiq’s optimizer only and nothing else. Listed below are the envisioned stages of introducing CBO in to Hive using Optiq:

  • Phase1 – Join Reordering & Join algorithm Selection
    • Table cardinality and boundary statistics will be used to compute operator cardinality.
    • Hive operator tree will be converted to Optiq operator tree.
    • Volcano optimizer in Optiq will be used to rearrange joins and to pick the join algorithm.
    • Optimized Optiq operator tree would be converted back to Hive AST and will be executed as before. So all of the Hive’s existing optimizations would run on top of Optiq optimized SQL.
  • Phase2 – Add support for Histograms, use other optimizations in Optiq
    • Introduce space efficient histograms
    • Change operator cardinality computation to use histograms
    • Register additional optimization rules available in Optiq like the ones listed above.
  • Phase3 – Code restructuring so that Optiq generates optimized Hive Operator tree
    • Unlike phase1 Hive AST would be directly translated into Optiq operator tree.
    • Optimize Optiq operator tree using Volcano optimizer.
    • Convert optimized Optiq operator tree back into Hive Operator tree. This is unlike phase1 where optimized Optiq operator tree is converted to Hive AST.

2. RELATED WORK

STATS

 

PAPERS

  • Query Optimization for Massively Parallel Data Processing

Sai Wu, Feng Li, Sharad Mehrotra, Beng Chin Ooi

School of Computing, National University of Singapore, Singapore, 117590

School of Information and Computer Science, University of California at Irvine

 

  • Profiling, What-if Analysis, and Cost-based Optimization of MapReduce Programs

Herodotos Herodotou Duke University, Shivnath Babu Duke University

 

  • Optimizing Joins in a Map-Reduce Environment

Foto N. Afrati, National Technical University of Athens, Greece

Jeffrey D. Ullman, Stanford University USA

 

  • Efficient and scalable statistics gathering for large databases in Oracle 11g

Sunil Chakkappen, Thierry Cruanes, Benoit Dageville, Linan Jiang, Uri Shaft, Hong Su, Mohamed Zait , Oracle Redwood Shores CA

http://dl.acm.org/citation.cfm?doid=1376616.1376721

 

  • Estimating Distinct (Postgress SQL)

http://wiki.postgresql.org/wiki/Estimating_Distinct

 

  • The History of Histograms

Yannis Ioannidis, Department of Informatics and Telecommunications, University of Athens

http://www.vldb.org/conf/2003/papers/S02P01.pdf

 

3. BACKGROUND

Hive Query optimization issues

Hive converts user specified SQL statement to AST that is then used to produce physical operator tree. All of the query optimizations are performed on the physical operator tree. Hive keeps semantic info separate from query operator tree. Semantic info is extracted during plan generation that is then looked up often during down stream query optimizations. Adding new query optimizations into Hive is often made difficult by the lack of proper logical query plan and due to Semantic info and query tree separation.

 

TEZ

Apache Tez generalizes the MapReduce paradigm to execute a complex DAG (directed acyclic graph) of tasks. Refer to the following link for more info.

http://hortonworks.com/blog/apache-tez-a-new-chapter-in-hadoop-data-processing/

Join algorithms in Hive

Hive only supports equi-Join currently. Hive Join algorithm can be any of the following:

Multi way Join

If multiple joins share the same driving side join key then all of those joins can be done in a single task.

Example: (R1 PR1.x=R2.a  - R2) PR1.x=R3.b - R3) PR1.x=R4.c - R4

All of the join can be done in the same reducer, since R1 will already be sorted based on join key x.

Common Join

Use Mappers to do the parallel sort of the tables on the join keys, which are then passed on to reducers. All of the tuples with same key is given to same reducer. A reducer may get tuples for more than one key. Key for tuple will also include table id, thus sorted output from two different tables with same key can be recognized. Reducers will merge the sorted stream to get join output.

Map Join

Useful for star schema joins, this joining algorithm keeps all of the small tables (dimension tables) in memory in all of the mappers and big table (fact table) is streamed over it in the mapper. This avoids shuffling cost that is inherent in Common-Join. For each of the small table (dimension table) a hash table would be created using join key as the hash table key.

 

Bucket Map Join

If the joining keys of map-join are bucketed then instead of keeping whole of small table (dimension table) in every mapper, only the matching buckets will be kept. This reduces the memory footprint of the map-join.

SMB Join

This is an optimization on Bucket Map Join; if data to be joined is already sorted on joining keys then hash table creation is avoided and instead a sort merge join algorithm is used.

Skew Join

If the distribution of data is skewed for some specific values, then join performance may suffer since some of the instances of join operators (reducers in map-reduce world) may get over loaded and others may get under utilized. On user hint, hive would rewrite a join query around skew value as union of joins.

 

Example R1 PR1.x=R2.a  - R2 with most data distributed around x=1 then this join may be rewritten as (R1 PR1.x=R2.a and PR1.x=1    - R2) union all (R1 PR1.x=R2.a and PR1.x<>1    - R2)

 

 

  • No labels