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

How to combine small files in Hadoop

Currently Hadoop is not built to work with a lot of small files. The following are the architectural limitations of hadoop that causes this problem

• In HDFS, all file metadata is stored in memory of the Namenode (which is most often a single big powerful machine). This means "more files=more memory". There is a limitation on the amount of memory you can add to a machine and that limits the amount of files that can be stored in Hadoop. 
• Namenode is used heavily for all jobs that run on Hadoop. More data in the memory can slow down Namenode and might end of slowing down the job execution time (it might be insignificant for long jobs though)
• There is a setup time required by Hadoop to run a mapper. By default, Hadoop will start minimum 1 mapper for every file in the input directory. Till Hadoop 0.20, hadoop does not lets us choose the number of mappers you want to run hence if your file is small, say 100K, then more time is wasted in Hadoop setup than actually processing the data.

There are couple different solutions to solve this problem

• Keep an eye on all data that is entered into HDFS  from other data sources. Try to optimize the processes, that push data to HDFS, to create files of size 128Mb (block size of HDFS). 
• If you have map reduce pipeline where output of a map reduce job become input of the next map reduce job then try to use reducers wisely in your jobs. If suppose your job uses 100 reducers and outputs files of size 10 MB each, and if the reducer computations are not CPU bound, then try to run the same job with less reducers (7-10). Remember - Hadoop creates one file for every reducer run even if reducer did not output any data. 
• If all else fails then try to combine small files into bigger files. Media6degrees has come up with a faily good solution to combine small files in Hadoop. You can use their jar straight out. See here for more details http://www.jointhegrid.com/hadoop_filecrush/index.jsp

Java templates/stubs for Mapper Reducer and Wrapper classes

A lot of times I want to test a concept in Hadoop that requires me to quickly create a small job and run it. Every job contains minimum 3 components

• Mapper Class
• Reducer Class
• Wrapper Class

The following are the templates I use to generate empty templates, just replace variable  <YOUNAME> with your class name 

-----------------------------------------------------------------------------------------------------------------------------------

MAPPER
-----------------------------------------------------------------------------------------------------------------------------------

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import java.io.IOException;

/* In case you are using Multiple outputs */

//import org.apache.hadoop.io.NullWritable;

//import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class <YOUNAME>Mapper extends Mapper<LongWritable, Text, Text, Text> {

    private Configuration conf;

    private Text outputKey = new Text();

    private Text outputValue = new Text();

    private String line = null;

    /* In case you are using Multiple outputs */

    //private NullWritable outputValue = NullWritable.get();

    //private MultipleOutputs<Text, Text> contextMulti = null;

    @Override

    public void setup(Mapper.Context context) {

        this.conf = context.getConfiguration();

        /* In case you are using Multiple outputs */

        //contextMulti = new MultipleOutputs<Text, Text>(context);

    }

    @Override

    public void map(LongWritable key, Text values, Context context)

            throws IOException, InterruptedException {

    }

    @Override

    public void cleanup (Mapper.Context context)throws IOException, InterruptedException {

        

        /* In case you are using Multiple outputs */

        //contextMulti.close();

    }

}

-----------------------------------------------------------------------------------------------------------------------------------

REDUCER
-----------------------------------------------------------------------------------------------------------------------------------

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/* In case you are using Multiple outputs */

//import org.apache.hadoop.io.NullWritable;

//import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public class <YOUNAME>Reducer extends Reducer<Text, Text, Text, Text> {

    private Configuration conf;

    private Text outputKey = new Text();

    private Text outputValue = new Text();

    private String line = null;

    /* In case you are using Multiple outputs */

    //private NullWritable outputValue = NullWritable.get();

    //private MultipleOutputs<Text, Text> contextMulti = null;

    @Override

    public void setup(Reducer.Context context) {

        this.conf = context.getConfiguration();

        /* In case you are using Multiple outputs */

        //contextMulti = new MultipleOutputs<Text, Text>(context);

    }

    @Override

    public void reduce(Text key, Iterable<Text> values, Context context)

            throws IOException, InterruptedException {

    }

    @Override

    public void cleanup(Reducer.Context context) {

        /* In case you are using Multiple outputs */

        //contextMulti.close();

    }

}

-----------------------------------------------------------------------------------------------------------------------------------

WRAPPER

This class uses following 2 classes 
https://sites.google.com/site/hadoopandhive/home/ExtendedFileUtil.java?attredirects=0&d=1
https://sites.google.com/site/hadoopandhive/home/StringUtil.java?attredirects=0&d=1
-----------------------------------------------------------------------------------------------------------------------------------

import StringUtil;

import ExtendedFileUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;


import java.io.IOException;
import java.text.ParseException;


public class <YOUNAME> extends Configured implements Tool, Constants {
    private Configuration conf = null;
    private Job job = null;
    private String inputDirList = null;
    private String outputDir = null;
    private String[] filesToProcess = null;
    private int totalReducers = 0;
    private int jobRes = 0;
    private ExtendedFileUtil fileUtil = new ExtendedFileUtil();


    public static void main(String[] args) throws Exception {
        <YOUNAME> ob = new <YOUNAME>();
        int jobRes = ToolRunner.run(ob, args);        
    }


    public int run(String[] args)
            throws ClassNotFoundException, IOException, InterruptedException, ParseException {
        jobRes = readCmdArgs(args);
        if (jobRes == 0) {
            jobRes = readConfig();
        }
        if (jobRes == 0) {
            jobRes = runMrJob();
        }
        return jobRes;
    }


    private int readCmdArgs(String[] args) {
        if (args.length == 2) {
            inputDirList = args[0];
            outputDir = args[1];
        } else {
            printUsage();
            System.exit(1);
        }
        return 0;
    }


    private int readConfig() throws IOException, InterruptedException, ClassNotFoundException {
        conf = new Configuration();
        //conf.set("SET_NEW_CONFIG_NAME", SET_NEW_CONFIG_VALUE);
        job = new Job(conf);
        if ((job.getJar() == null) || (job.getJar() == "")) {
            job.setJarByClass(<YOUNAME>.class);
        }
        return 0;
    }


    private int runMrJob()
            throws IOException, InterruptedException, ClassNotFoundException {
        filesToProcess = fileUtil.getFilesOnly(inputDirList, true);
        job.setJobName("<YOUNAME>");
        TextInputFormat.addInputPaths(job, StringUtil.arrayToString(filesToProcess, ","));
        TextOutputFormat.setOutputPath(job, new Path(outputDir));
        System.out.println("Input Dir: " + inputDirList);
        System.out.println("Output Dir: " + outputDir);


        job.setMapperClass(<YOUNAME>Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);


        job.setReducerClass(<YOUNAME>Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        totalReducers = Math.round((fileUtil.size(inputDirList) / 134217728) * 0.1F);
        totalReducers = Math.max(totalReducers, 1);
        job.setNumReduceTasks(totalReducers );
        deleteOutputDirectory(outputDir);
        jobRes = job.waitForCompletion(true) ? 0 : 1;
        deleteLogsDirectory();
        fileUtil.removeAllZeroByteFiles(outputDir);
        return 0;
    }




    private int deleteOutputDirectory(String outputDir) throws IOException {
        fileUtil.removeHdfsPath(new Path(outputDir).toString());
        return 0;
    }


    private int printUsage() {
        System.out.println("USAGE: <YOUNAME> <inputDirList> <outputDir>");
        return 0;
    }


    private int deleteLogsDirectory()
            throws IOException {
        Path outputLogPath = new Path(new Path(outputDir).toString() + "/" + "_logs");
        fileUtil.removeHdfsPath(outputLogPath.toString());
        return 0;
    }
}

How configure Secondary namenode on a separate machine

If you have installed cloudera's hadoop distribution (CDH2) then you must have noticed that running command start-dfs.sh starts an instance of SecondaryNameNode process on all the datanodes. This is happening due to the way SecondaryNameNode startup is defined in file bin/start-dfs.sh. 


Scenario 1 : If you want to run your SecondaryNameNode on some other server (say sn.jeka.com) instead of the datanodes then do the following 

1. Logon to JobTracker (I am going to JobTracker because I have set variable HADOOP_MASTER in file ${HADOOP_HOME}/conf/hadoop-env.sh to point to the JobTracker hence any changes made there will be synched to your cluster) 

• Create a new file ${HADOOP_HOME}/conf/secondarynamenode and add following line
  sn.jeka.com
• In file ${HADOOP_HOME}/bin/start-dfs.sh, replace line
  "$bin"/hadoop-daemons.sh --config $HADOOP_CONF_DIR --hosts masters start secondarynamenode
  with
  ssh $(cat $HADOOP_CONF_DIR/secondarynamenode) "${bin}/hadoop-daemon.sh --config $HADOOP_CONF_DIR --hosts secondarynamenode start secondarynamenode;exit"
• In file ${HADOOP_HOME}/bin/stop-dfs.sh, replace line
  "$bin"/hadoop-daemons.sh --config $HADOOP_CONF_DIR --hosts masters stop secondarynamenode
  with
  ssh $(cat $HADOOP_CONF_DIR/secondarynamenode) "${bin}/hadoop-daemon.sh --config $HADOOP_CONF_DIR --hosts secondarynamenode stop secondarynamenode;exit"

2.  Logon to Namenode and execute the following commands

• ${HADOOP_HOME}/bin/stop-dfs.sh; ${HADOOP_HOME}/bin/start-dfs.sh; ${HADOOP_HOME}/bin/stop-dfs.sh; ${HADOOP_HOME}/bin/start-dfs.sh

You have to start and stop twice because in the first start, the code will be synched from JobTracker 

Thats! it. You secondary name node process will now start on the designated server, i.e. sn.jeka.com and not on the datanodes. 

Scenario 2 : If you want to run your SecondaryNameNode on the NameNode (say nn.jeka.com) itself then do the following 

Follow same steps as Scenario 1 except that replace all intances of sn.jeka.com to nn.jeka.com

Scenario 3 : If you do not want to run secondary name node at all then do the following

Follow same steps as Scenario 1 except that instead of replacing lines, delete them. 

How to control a Hadoop job using the web interfaces provided by the Job Tracker and Name Node

Hadoop provides a great way to manage your jobs and operating on HDFS using the web interface by setting the property webinterface.private.actions to true in file src/core/core-default.xml. 


When set to true, the web interfaces of JobTracker and NameNode may contain  actions, such as kill job, delete file, etc., that should  not be exposed to public.

Note: Enable this option only if the web interfaces for JobTracker and Name node are reachable by those who have the right authorizations. 

<property>

  <name>webinterface.private.actions</name>

  <value>false</value>

</property>