Hadoop Cluster size deployed by CBS Interactive

Following are some details of the Hadoop Production Infrastructure deployed at CBS Interactive as of now
- Total Nodes: 80
- Total Disk Capacity: 1 PB
- ETL apps were written using Hadoop Streaming (Phython)

How, in hive, to create a column name that is same as a reserved keyword used by Hive

If you run the follwoing command in hive, it will fail because sort is a reserved keyword in hive

CREATE EXTERNAL TABLE aaaabc ( sort STRING  )
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\002'
LINES TERMINATED BY '\n';

FAILED: Parse Error: line 1:31 mismatched input 'sort' expecting Identifier in column specification


To ovecome this, run the command by putting sort in backticks, i.e.
CREATE EXTERNAL TABLE aaaabc ( `sort` STRING  )
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\002'
LINES TERMINATED BY '\n';

Hive Error: FAILED: Parse Error: line 54:4 mismatched input expecting Identifier in column specification

PROBLEM
I was trying to create an external table in hive using the following command
CREATE EXTERNAL TABLE tetl_fact_r
(
    custid STRING,
    value STRING,
    ph STRING,
    email STRING,
    sort STRING,
    address STRING,
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY '\002'
LINES TERMINATED BY '\n'
STORED AS TEXTFILE
 LOCATION '/user/hadoop-blog/abc.text';

and got the following error

FAILED: Parse Error: line <linenum>:4 mismatched input <column name> expecting Identifier in column specification

REASON
From https://issues.cloudera.org/browse/SQOOP-37, I found that this is happening because column name sort is a reserved keyword in hive.

SOLUTION
To be honest, I din't spend much time finding the solution, instead I just renamed my field and it worked after that. Hence the new query looked like
CREATE EXTERNAL TABLE tetl_fact_r
(
    custid STRING,
    value STRING,
    ph STRING,
    email STRING,
    sorttype STRING,
    address STRING,
)

If any of you figure out a way to make this query work without renaming the column then please leave a comment. It will be much appreciated.

How to dynamically assign reducers to a Hadoop Job in runtime

When we were working on setting our production jobs we came to a point where we needed a way to dynamically assign reducers to a Hadoop Job in runtime. Following is what we have implemented as of today.  If you are reading the blog and have a better idea to share then please leave a comment.

PROBLEM STATEMENT

We have a lot of cases where same produciton job works on different sizes of data sets, i.e. for an hour the job can process from 35GB to 400Gb data. We wanted to change the number of reducers depending on the data sets, in runtime, and also we did not want a job to hog all reducers because the grid was shared between jobs from different teams. All jobs were equally important.

SOLUTION

This is something that hadoop MR framework cannot do for you. We found that hive came up with a solution by providing a property that limits the maximum number of bytes that will be processed by one reducer hence if this property is set to 1GB and data size is 1.2 GB data then 2 reducers will be assigned to the job in runtime; if data size 10 GB data then 10 reducers will be assigned to the job in runtime.  This works great for hive because it is designed to work agnostic of the dataset. Also a big job in hive can still take all the reducers in the grid. Since we knew our data very well and also did not want a single job to take all reducers hence we decided to implement our own solution that solved our problem.

The solution is very simple. We asked all the job owners to run their jobs in the sandbox on a fixed set of data and provide us either of the following

1.       Required: Input size of the data in Mb

2.       Required: Output size of the data in Mb

3.       Required: Is the reducer calculation CPU bound or I/O bound

4.       Optional: A decimal number, the multiplier,  to fine tune the number of reducers (if not provided then 1 will be used)

OR

1.       Required: Provide fixed number of reducers (they have to be less than TotalReducersOnTheGrid/2)

For CPU bound jobs, total number of reducers were calculated as

Total Reducers = Minimum((InputSizeOfDataInMb/128Mb) * multiplier, TotalReducersOnTheGrid/2)

i.e. total number of reducers should be equal to either the Input data size divided by the HDFS block size multiplied by the multiplier or half the total number of reducers available in the grid, whichever is smaller.

For I/O bound jobs the total number of reducers were calculated using the following formula

Total Reducers = Minimum((OutputDataSize/InputDataSize) * (InputSizeOfDataInMb/128Mb) * multiplier, TotalReducersOnTheGrid/2)

The concept of multiplier was introduced to optimize the jobs when the generic formula was not enough to optimize the number of reducers for the job. We found that some jobs always required an exact number of reducers regardless of the size of data set hence we also provided the job owners a way to specify that.

This pretty much solved most of our problems.

Hadoop cluster at Ebay

I am always curious to know how other companies are installing Hadoop clusters. How are they using its ecosystem. Since Hadoop is still relatively new, there are no best practices. Every company is implementing what they think is the best infrastructure for the Hadoop Cluster.

At Hadoop NYC 2010 conference, ebay showcased there implementation of Hadoop production cluster. Following are some tidbits on ebay's implementation of Hadoop.

- JobTracker, Namenode, Zookeeper, HBase Master are all enterprise nodes running in Sun 64 bit architecture. They are running red hat linux with 72GB Ram and 4TB disks.
- There are 4000 datanodes, each running cent OS with 48 GB RAM and 10TB space
- Ganglia and Nagios are used for monitoring and alerting. Ebay is also building a custom solution to augment them.
- ETL is done using mostly Java Map Reduce programs
- Pig is used to build data pipelines
- Hive is used for AdHoc queries
- Mahout is used for Data Mining

They are toying with the idea of using Oozie to manage work flows but haven't decided to use it yet.

It looks like they are doing all the right things.

ERROR: hdfs.DFSClient: Exception in createBlockOutputStream java.io.IOException: Bad connect ack with firstBadLink

While running a job once I got the following exception


10/12/10 21:09:05 INFO hdfs.DFSClient: Exception in createBlockOutputStream java.io.IOException: Bad connect ack with firstBadLink 10.1.73.148:50010
10/12/10 21:09:05 INFO hdfs.DFSClient: Abandoning block blk_3623545154924652323_87440
10/12/10 21:09:11 INFO hdfs.DFSClient: Exception in createBlockOutputStream java.net.ConnectException: Connection refused
10/12/10 21:09:11 INFO hdfs.DFSClient: Abandoning block blk_-4726571439643867938_87441\


REASON
The error contains the IP address (10.1.73.148) of the tasktracker/datanode machine for which the exception is thrown. The exception is thrown because the datanode daemon is not running on that machine; you can check this by logging into this machine, lets use 10.1.73.148 in the example, and running command
ps -eaf | grep "DataNode" | grep -v "grep"
If no lines are returned then this means that datanode daemon is not running on 10.1.73.148.

What happened is that machine 10.1.73.148 contain a data block that is required for the job that you are trying to run. If this block is replicated on other machines and those machines are running datanode daemons then this is not a problem, Hadoop will get the data block from some other machine and continue the job but if for any reason the data block is not available on any other node then your job will fail.


RESOLUTION
Logon to 10.1.73.148 and run the following command
hadoop-daemon.sh start datanode
The above command should start the datanode daemon on 10.1.73.148. You can double check this my running command
ps -eaf | grep "DataNode" | grep -v "grep"
It should return 1 line

Thats it. Try running the job again. It should not throw exception anymore

How to see table definition (extended) in Hive

To see table definition in Hive, run command
describe table name;

To see more detailed information about the table, run command
describe extended [tablename];

ERROR: java.lang.IllegalArgumentException: Name cannot be have a '' char

ERROR
Sometimes your Hadoop MapReduce job can fail with the following exception
java.lang.IllegalArgumentException: Name cannot be have a '' char
at org.apache.hadoop.mapreduce.lib.output.MultipleOutputs.checkTokenName(MultipleOutputs.java:149)
at org.apache.hadoop.mapreduce.lib.output.MultipleOutputs.checkNamedOutputName(MultipleOutputs.java:175)
at org.apache.hadoop.mapreduce.lib.output.MultipleOutputs.write(MultipleOutputs.java:352)
at org.apache.hadoop.mapreduce.lib.output.MultipleOutputs.write(MultipleOutputs.java:337)
at learnhadoop.MosMapper.map(MosMapper.java:38)
at learnhadoop.MosMapper.map(MosMapper.java:14)
at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:583)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:305)
at org.apache.hadoop.mapred.Child.main(Child.java:170)


REASON
This happens when you use MultipleOutputs class in your Hadoop Job and try to name the output file with non-alphanumeric characters (like : or - etc.)
i.e. <MultipleOutput Object>.write(KEY,VALUE,"Fruit::Mango") will throw this error because you are using colons in the output file name



RESOLUTION
Try to use only alphanumeric characters in the output filename and if you absolutely have to use some kind of delimiter, stick with dot (.)
i.e. <MultipleOutput Object>.write(KEY,VALUE,"Fruit..Mango")  will not throw this error

MultipleOutputs performance issues

MultipleOutputs class in Hadoop API provides a very neat way of separating disparate data but it comes with a performance hit.

I found that some of my production jobs slowed down after I refactored by code to use MultipleOutputs class. I did some benchmarking to ensure that its not the cluster but MultipleOutputs class that slowed my processes down.

I setup a small cluster with just 6 machines and some data

• 1 machine running JobTracker
• 1 machine running Namenode
• 4 machines running Datanodes and tasktracker
• Input data 8Gb

All machines were of same size and nothing else was running on them during benchmarking. 

Test 1: Mapper without MultipleOutputs

I created a mapper that 

• Reads a file line by line
• Creates output file name on the fly by taking first 3 characters of the hash of the input line. This information was not used to write output (because we are not using MultipleOutputs yet).  
• Write the output key as input line and outputValue as NullWritable

I ran it 5 times and the median runtime was 4m 40s. 

Test 2: MultipleOutputs Mapper

Then I modified the above mapper to use the output file name and write data out using MultipleOutputs. I ran this 5 times and the median runtimes was 5m 48s. 

Based on this benchmark I found that MultipleOutputs slows down a job by almost 20%. 

This happens because more small files are created when you use MultipleOutputs class.

Say you have 50 mappers then assuming that you don't have skewed data, Test1 will always generate exactly 50 files but Test2 will generate somewhere between 50 to 1000 files (50Mappers x 20TotalPartitionsPossible) and this causes a performance hit in I/O. In my benchmark, 199 output files were generated for Test1 and 4569 output files were generated for Test2. 

Extended FileUtil class for Hadoop

While writing production jobs in Hadoop I identified following tasks that were required for some MapReduce jobs but were not readily available in Hadoop 0.20 API

1. Get size of a file or directory in HDFS
• We require this to dynamically change the number of reducers used for a job by looking at the amount of input data that the job will process
2. Recursively remove all zero byte files from a directory in HDFS. 
• This happens a lot when you use MultipleOutput class in reducer (impact is less when used in Mapper). A lot of times the reducer does not gets any record for which a MutipleOutput file needs to be created hence it creates a 0 byte files. These files have no use, its best to remove them after the job is finished. 
3. Recursively get all subdirectories of a directories 
4. Recursively get all files within a directory and its sub directories
• By default, as of now, when Hadoop job is run, it only processes the immediate files under the input directory, any files in the subdirectories of the input path are not processed hence if you want your job to process all files under the subdirectories also then its better to create a comma delimited list of all files within the input path and submit it to the job.  

All the above tasks were implemented in the ExtendedFileUtil class. Source code can be found at
https://sites.google.com/site/hadoopandhive/home/ExtendedFileUtil.java?attredirects=0&d=1

The wrapper class on link http://hadoop-blog.blogspot.com/2010/12/java-templatesstubs-for-mapper-reducer.html contains an example of how to use ExtendedFileUtil class