Hadoop Performance tuning (Hadoop-Hive) Part 2

[Note: This post is second part of Hadoop performance tuning, if you directly reached this page, please click here for part 1.]

I am testing these parameters with Hadoop and Hive framework from sql based queries. For checking performance improvement with configuration parameters, I use sample data of 100 million records and running some complex queries in Hive interface in top of Hadoop. In this part 2 we will see few more Hadoop configuration parameter to get maximum performance improvement in Hadoop cluster. 

Map Output compression ( mapred.compress.map.output )

By default this value set to false, its recommend to set this parameter to true for cluster with large amount of input data to be processed.  Because of compression data transfer between nodes are fast. Map output will not directly move to reducer, intermediately it will write to disk. So this setting helps to save disk space and fast disk read/write. And it’s not recommended to set this parameter to true for small amount of input data to be processed, because it will increase the processing time for compressing and decompressing data. But for Big data compressing and decompression time is considerably small when compare to time its saves in transferring and disk read/write. 

Once we set above configuration parameter to true, other dependent parameter will be active such as setting compression technique (codec) and compression type.  

Compression method or technique orcodec (mapred.map.output.compression.codec )

Default value for this parameter is org.apache.hadoop.io.compress.DefaultCodec. Other available codec are org.apache.hadoop.io.compress.GzipCodec. DefaultCodec will take more time but more compression. In LZO method it will take less time for compression amount of compression is less. Our own codec also can be added. Add codec or compression library which is suitable (best) for your input data type. 

mapred.map.output.compression.type parameter help to identify in which basis data should be compressed. User can set either RECORD or BLOCK. Record type is default type in which each individual value is compressed, means it will compress whole data block as it is. Block type is recommended one, in which data compressed based on data block key-value pairs, so it helps for sorting data in reducer side. In Cloudera Hadoop, default type is set to Block for better performance.

Three more configuration parameters are there 

1. mapred.output.compress

2. mapred.output.compression.type 

3. mapred.output.compression.codec

Same above rules apply here, but this parameter meant for MapReduce job output, first three parameters specify compressed output for map output alone. These three configuration parameter specify for all job output which should be compressed or not and in which type and codec.

Above suggestions are observed with Hadoop cluster with Hive querying, please leave a comment and recommend this post by clicking  Facebook ‘Like’ button and ‘+1’ at bottom of this page.

Hadoop with Hive

Nowadays, there are lots of Hadoop emerging. Indeed, by “Lots of Hadoop”, I mean companies releasing their own versions of Hadoop (e.g. Cloudera) by building a layer over the original Apache Hadoop distribution. We can also call these “customized” versions of Apache Hadoop. But when we think about the core part, it remains the same across different Hadoop flavors. Apache Software Foundation (ASF) focuses on improving Hadoop by bringing many smaller sub-projects under it to facilitate open source tools development around Hadoop. Hive happens to be one of Hadoop’s more prominent child projects.

Hive is a data warehouse infrastructure, initially developed by Facebook. Hadoop with Hive combination gives us advantages of Distributed File System, Map-Reduce and SQL. As we know, to process huge amounts of data in Hadoop for each and every process/operation, we have to write new Map-Reduce program (job). For users with limited number of operations or sequences of same operation, this task will be an easy one. But for those whose requirements are a bit more prone to change, the challenge is they have to write new Map-Reduce program for every new requirement. Unfortunately, this is the only way to deal with unstructured data.

But for structured data, like logging (log4j) files, relational type data, and other similar, more predictable sets, the data can be stored in table-like structures. This is the area where Apache Hive really shines. Hive is a layer running on top of Hadoop that helps process the data in Hadoop by using SQL-like queries written in Hive Query Language (HQL). While loading data in HDFS through Hive as table, it also stores metadata of input, which describes the structure of input data. Note that Hive is required to be installed on the Hadoop master node. Hive converts an input query into a Map-Reduce job and submits it to Hadoop, making it easy for users to analyze and process data.

Hive Prerequisites

• Hadoop 0.20 and above
• Java 1.6 and above
• MySQL or Derby lightweight database in master node to store only Hive metadata

Advantages of Hive

• Supports rich data types like List, Map, and Struct, apart from basic data types.
• Provides Web UI and Command Line Interface UI that are incorporated for querying data. This provides helpful tools for developers and learners for testing and debuging their queries.
• Thrift server that comes with Hive helps with JDBC and ODBC connections, so any application can interact with Hive to Hadoop as a backend database. Thrift takes care of language conversion, which allows ANY type of language program to interact with Hadoop.
• Even for complex structured input data, we can write our own DeSer (serializers and deserializers) programs for parsing input data, storing their table structure in metadata repository, and loading data on Hadoop File System (HDFS).
• Supports queries with SQL filters, Joins, Group By, Order By, Inner Table, Functions, and other SQL-like operators. Using HQL we can also redirect query output to a new table. Along with all SQL features, we can also attach our own functions and Map-Reduce programs as the part of HQL query.
• Partition and Bucket: partitioning helps split data into different chunks based on input value range, which allows to skip unwanted data while executing queries. Bucket split data is based on a hash function. Both help to improve the performance of querying.
• Optimizers are being developed by Apache for Hive for better performance. We can improve our Hadoop and Hive performance by tuning few configuration parameters based on our application requirements. To learn more, read my recent article on Hadoop and Hive Performance Tuning.
• Hive is used by major companies like Facebook, Yahoo, and Amazon. Hadoop and Hive play a major role in the proliferation of Cloud Computing. Amazon provides S3 (Simple Storage Service) and Elastic MapReduce as a service in cloud environment, which is a Cloud server pre-installed with Hadoop and Hive. It allows us to load our data in Hadoop (Elastic MapReduce) and execute queries on it with the help of Hive. Amazon Elastic MapReduce is a successful product which uses Hadoop and Hive jointly. Click here to learn more about how this technology works.

With more and more Hadoop distributions appearing in the “wild”, it’s clear that this project isn’t going anywhere anytime soon. If anything, it will only gain momentum as more and more companies switch to Hadoop to handle their large data repositories. Hive is a relatively mature Hadoop sub-project companion that facilitates easy data analysis, ad-hoc queries, and manipulation of large datasets stored in Hadoop. These two are a “Match Made in Heaven”!

Above suggestions are observed with Hadoop cluster with Hive querying, please leave a comment and recommend this post by clicking  Facebook ‘Like’ button and ‘+1’ at bottom of this page.

Hadoop Performance Tuning (Hadoop-Hive)

Hadoop Cluster performance tuning is little hectic, because hadoop framework uses all type of resource for processing and analyzing data. So tuning its parameter for good performance is not static one. Parameter values should be change based on clusters following items for better performance:

• ·         Operating System
• ·         Processor and its number of cores
• ·         Memory (RAM)
• ·         Number of nodes in cluster
• ·         Storage capacity of each node
• ·         Network bandwidth
• ·         Amount of input data
• ·         Number of jobs in business logic

Recommended OS for hadoop clusters is Linux, because windows and other GUI based OS runs lot of GUI (Graphical user interface) processes and will occupy most of the memory.

Storage capacity of each node should have at-least 5GB extra after storing distributed HDFS input data. For Example if input data in 1 TB and with 1000 node cluster means, (1024GB x 3(replication factor))/1000 nodes = approx 3GB of distributed data in each node, so it is recommended to have at-least 8GB of storage in each node. Because each data node writes log and need some space for swapping memory.

Network bandwidth is recommended to have at-least 100 Mbps, as well known while processing and loading data into HDFS, Hadoop moves lot of data over network. Lower bandwidth channel also degrade the performance of hadoop cluster.

Number of nodes requires for cluster is depends on amount of data to be processed and capacity of each node. For example node with 2GB Memory and 2 core processor can process 1GB of data in average time. It can also process 2 data block (of 256MB 0r 512MB) simultaneously. For Example:  To process 5TB of data, it is recommended to have 1000 nodes with 4-to-8 Core processor and 8-to-10 GB of memory in each node to produce result in few minutes.

Hadoop Parameters:

Data block size (Chunk size): 

        dfs.block.size parameter will be in hdfs-site.xml file, parameter value is mentioned in number of bytes. Block size should be chosen completely based on each node memory capacity. If memory is less then set smaller block size. Because TaskTracker, bring whole block of data to memory while processing. So for 512MB RAM, it is advised to set block size as 64MB or 128MB. If it is dual core processor then TaskTracker can process 2 block of data at same time, so two data block will be bring to memory while processing, so it should be planned according to that, for this have to set concurrent tasktracker parameter also.

Number of Maps and Reducer:

           mapred.reduce.tasks & mapred.map.tasks parameter will be in mapred-site.xml file. By default, number of maps will be equal to number of data block. For example, if input data is 2GB and block size is 256MB means, while processing 8 Maps will run. It won’t bother about memory capacity and number of processor. So we need to tune this parameter to number of nodes*number of cores in each node.

Number of Maps = Total number of processor core available in cluster.

As per above example it runs 8 Maps, if that cluster have only 4 processor core, then multiple thread will start running and keep swapping the memory data, which will degrade the performance of hadoop cluster. In same way set number of reducer to number of core in cluster. After mapping job is over, most of nodes go idle and few nodes working for reducer to complete, to make reducer job to complete fast, set its value to number of nodes or number of core processor.

Logging Level:

            HADOOP_ROOT_LOGGER = ERROR set this value in hadoop script file. By default its set to INFO mode, in information mode, hadoop will log all information about including all event, jobs, tasks completed, IO info, warning and error. It won’t increase huge performance improvement, but it will help to reduce number of log file I/Os and give small improvement in performance.

Above suggestions are observed with Hadoop cluster with Hive querying, please leave a comment and recommend this post by clicking  Facebook ‘Like’ button and ‘+1’ at bottom of this page.

Optimizing Joins in hive/Sorting Java Heap issues with hive joins

In hadoop we tent to use hive extensively since it is SQL like language and easier in framing our jobs with stored structured data. (Even Pig is great but still needs a little time to get comfortable with Pig Latin). But as beginners we often get struck with hive joins in large data sets. It is a common scenario running into out of memory/java heap space errors on joins with huge hive tables. We can avoid these bottlenecks to a greater extent utilizing a few smarter options available with hive. Let us look into a few of them.

1.       Enable map joins

It is a pretty good approach to enable map joins in hive when you are trying to do a join with multiple tables and if one or more of them has a smaller data volume. With this enabled the smaller tables would be distributed on the distributed cache as a hash table by a local map reduce task before the actual map reduce job. This could save considerable time as it turns to be map side join compared to running a common map reduce side join (normal hive join). You need to set the following at hive CLI before running the join Query

               set hive.auto.convert.join = true;

The point to be noted here is that, hive is intelligent enough with map side joins and if the data volume is larger not to fit into map side joins it executes the backup task, ie the common  join(full map reduce execution) to accomplish the job. So when you are taking performance into consideration the time to check on the executablity of map  join is an overhead, so if you are sure the data in the tables that you try to join is always huge then better not enabling the same for your job.

We were mentioning the term ‘small tables’ a lot here. But how small this table has to be? By default the small table size is 25 Mb. So if the table is larger than 25 Mb then the hive common join would be triggered. However 25Mb is conservative and you can modify the same to a desirable value by setting the following configuration variable.

                set hive.smalltable.filesize = 40000000;

2.       Exploiting EQUI Join support in hive

Hive supports only Equi joins and we need to exploit the same in our hive query to get rid of OOM errors as well another common scenario of hive queries running infinitely. It is relatively straight forward, if your join query has a few where clauses with equality then include them inside the ON clause in your joins. It’d considerably reduce the number of records chosen to join and hence lesser number of records in sort phase.

                For eg: let us consider a query like this

Select Table1.Column1, Table2.Column2 FROM Table1 JOIN Table2 ON (Table1.Column5 = Table2.Column7 AND Table1.Column9=Table2.Column3) WHERE Table1.Column1 = ‘1024’ AND Table2.Column2 > 5;

This Query has an equality expression in the where clause involving one of the tables in join, we can optimize our hive query by including the equality filter condition as part of join as

Select Table1.Column1, Table2.Column2 FROM Table1 JOIN Table2 ON (Table1.Column5 = Table2.Column7 AND Table1.Column9=Table2.Column3 AND Table1.Column1 = ‘1024’) WHERE Table2.Column2 > 5;

The difference it creates in execution is that in first query after the join the filter condition is applied which means there would be more records involved in join but in the second query the filter condition is done before/on joining hence less records in join.

3.       Increase the heap Size

Definitely this has to be one of the options if your hive query is already optimized and satisfies the first two checks and still the execution halts due to heap size issues. You can increase the heap size for the map reduce child tasks by setting the property ‘mapred.child.java.opts’ to a higher value.  Like for 1GB set it as

mapred.child.java.opts =  -Xmx1024m

Definitely there are many other options to deal with these issues like working on io.sort.mb deciding on the maximum number of mappers/reducers etc. Left up to your choice to google on and hit the bulls eye based on your use case/hive query

Performance tuning of hive queries

Hive performance optimization is a larger topic on its own and is very specific to the queries you are using. Infact each query in a query file needs separate performance tuning to get the most robust results.

I'll try to list a few approaches in general used for performance optimizationLimit the data flow down the queries

When you are on a hive query the volume of data that flows each level down is the factor that decides performance. So if you are executing a script that contains a sequence of hive QL, make sure that the data filtration happens on the first few stages rather than bringing unwanted data to bottom. This will give you significant performance numbers as the queries down the lane will have very less data to crunch on.

This is a common bottle neck when some existing SQL jobs are ported to hive, we just try to execute the same sequence of SQL steps in hive as well which becomes a bottle neck on the performance. Understand the requirement or the existing SQL script and design your hive job considering data flow

Use hive merge files

Hive queries are parsed into map only and map reduce job. In a hive script there will lots of hive queries. Assume one of your queries is parsed to a mapreduce job and the output files from the job are very small, say 10 mb. In such a case the subsequent query that consumes this data may generate more number of map tasks and would be inefficient. If you have more jobs on the same data set then all the jobs will get inefficient. In such scenarios if you enable merge files in hive, the first query would run a merge job at the end there by merging small files into  larger ones. This is controlled
using the following parameters

hive.merge.mapredfiles=true
hive.merge.mapfiles=true (true by default in hive)

For more control over merge files you can tweak these properties as well

hive.merge.size.per.task (the max final size of a file after the merge task)

hive.merge.smallfiles.avgsize (the merge job is triggered only if the average output filesizes is less than the specified value)

The default values for the above properties are

hive.merge.size.per.task=256000000

hive.merge.smallfiles.avgsize=16000000

When you enable merge an extra map only job is triggered, whether this job gets you anoptimization or an over head is totally dependent on your use case or the queries.

Join Optimizations

Joins are very expensive.Avoid it if possible. If it is required try to use join optimizations as map joins, bucketed map joins etc

http://venkatarun-n.blogspot.in/2012/12/optimizing-joins-in-hivesorting-java.html

There is still more left on hive query performance optimization, take this post as the baby step. More tobe added on to this post and will be addded soon . :)