Geo IP client

Sometimes you might want to find out location of where your web traffic is coming from. In that case you might want to look into the access log generated by your web server, it will give you ip address of the client. ONce you have the client's ip you can use GeoIP database to look up city, state, country for that ip. You can download the sample project from here This is how a sample log line looks like

125.215.163.73 - - [29/Aug/2011:00:00:18 -0700] "GET /blog/geekery/xvfb-firefox.html HTTP/1.1" 200 9767 "http://www.google.com/url?sa=t&source=web&cd=1&ved=0CBkQFjAA&url=http%3A%2F%2Fwww.semicomplete.com%2Fblog%2Fgeekery%2Fxvfb-firefox.html&rct=j&q=xvfb%20firefox%20fully%20loaded%20screenshot&ei=zDhbTvX6N-jmiAK5_7i4CQ&usg=AFQjCNEFaYxjYoKmLd5CLaG3SbQNStGkLg&sig2=oZnDDXKzFB8uwbDg5aNi2w&cad=rja" "Mozilla/5.0 (Windows; U; Windows NT 6.1; en-US; rv:1.9.2.20) Gecko/20110803 Firefox/3.6.20 FirePHP/0.5"

Now you can download GeoIP database from MaxMind, then create a simple Java Client that given a ip address will print its geo location like this, you can download source code for my project from here

	package com.spnotes;
	import java.io.File;
	import java.io.IOException;
	import java.net.InetAddress;
	import com.maxmind.geoip2.DatabaseReader;
	import com.maxmind.geoip2.exception.GeoIp2Exception;
	import com.maxmind.geoip2.model.CityResponse;
	/**
	* Created by gpzpati on 2/10/14.
	*/
	public class GeoIP {
	private void getIPLocation(String ipStr) throws IOException, GeoIp2Exception{
	File file = new File("/temp/data/GeoLite2-City.mmdb");
	DatabaseReader reader = new DatabaseReader.Builder(file).build();
	CityResponse response = reader.city(InetAddress.getByName(ipStr));
	System.out.println("City " +response.getCity());
	System.out.println("ZIP Code " +response.getPostal().getCode());
	System.out.println("Country " +response.getCountry());
	System.out.println("Location " +response.getLocation());
	}
	public static void main(String[] argv)throws Exception {
	GeoIP gih = new GeoIP();
	gih.getIPLocation("210.128.115.149");
	gih.getIPLocation("67.188.26.1");
	}
	}

view raw GeoIP.java hosted with ❤ by GitHub

Your output should look something like this

City Tokyo
ZIP Code null
Country Japan
Location Location [latitude=35.685, longitude=139.7514, timeZone=Asia/Tokyo]
City Fremont
ZIP Code 94536
Country United States
Location Location [latitude=37.567, longitude=-121.9829, metroCode=807, timeZone=America/Los_Angeles]

Connecting to Hive using Beeline client

I like to use Hive Beeline client better than using the default hive command line interface. On my cloudera VM 4.4.0-1 i can connect to beeline client by executing following command beeline -u jdbc:hive://

Connecting to Hive using JDBC client

I wanted to try out connecting to Hive using a JDBC driver, so i followed these steps. You can download the Maven project from this location

1. First start the HiveServer by executing hive --service hiveserver command
   
3. Make sure that Hive is running by executing netstat --tnlp | grep 10000
   
4. Now use the following Java code for connecting and executing simple select query on Hive

   	package com.spnotes.hive;
   	import java.sql.Connection;
   	import java.sql.DriverManager;
   	import java.sql.ResultSet;
   	/**
   	* Created by gpzpati on 4/26/14.
   	*/
   	public class HelloHiveJDBC {
   	private static String driverName = "org.apache.hadoop.hive.jdbc.HiveDriver";
   	public static void main(String[] argv) throws Exception{
   	try {
   	Class.forName(driverName);
   	} catch (ClassNotFoundException e) {
   	// TODO Auto-generated catch block
   	e.printStackTrace();
   	System.exit(1);
   	}
   	Connection connection = null;
   	try {
   	connection= DriverManager.getConnection("jdbc:hive://172.16.225.188:10000/default", "", "");
   	ResultSet resultSet = connection.createStatement().executeQuery("select * from records");
   	while (resultSet.next()) {
   	System.out.println(resultSet.getObject("year") + " " + resultSet.getObject("temperature"));
   	}
   	}finally{
   	connection.close();
   	}
   	}
   	}

   view raw HelloHiveJDBC.java hosted with ❤ by GitHub

MapReduce program that reads input files from S3 and writes output to S3

In the WordCount(HelloWorld) MapReduce program entry i talked about how to create a simple WordCount Map Reducer program with Hadoop. I wanted to change it to so that it reads input files from Amazon S3 bucket and writes output back to Amazon S3 bucket, so i built S3MapReduce program, that you can download from here. I followed these steps

1. First create 2 buckets one for storing input and other for storing output in your Amazon S3 account. Most important issue here is to make sure that you create your buckets in US Standard region, if you dont do that then additional steps might be required for Hadoop to be able to access your buckets Name of input bucket in my case is com.spnotes.hadoop.wordcount.books
   
   Name of the output bucket is com.spnotes.hadoop.wordcount.output
   
2. Upload few .txt files that you want to use as input in your input bucket like this
   
3. Next step is to create MapReduce program like this, In my case one Java class has code for Mapper, Reducer and driver class. Most of the code in the MapReduce is same only difference is for working with S3 you will have to add few S3 specific properties like this, basically you need to set your accessKey and secretAccessKey that you can get from AWS Security console and paste it here. You will also have to tell Hadoop to use s3n as file system.

   
   //Replace this value
   job.getConfiguration().set("fs.s3n.awsAccessKeyId", "awsaccesskey");
   //Replace this value
   job.getConfiguration().set("fs.s3n.awsSecretAccessKey","awssecretaccesskey");
   job.getConfiguration().set("fs.default.name","s3n://com.spnotes.hadoop.input.books");
   

   	package com.spnotes.hadoop;
   	import org.apache.hadoop.conf.Configured;
   	import org.apache.hadoop.fs.Path;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.LongWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Job;
   	import org.apache.hadoop.mapreduce.Mapper;
   	import org.apache.hadoop.mapreduce.Reducer;
   	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   	import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
   	import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
   	import org.apache.hadoop.util.Tool;
   	import org.apache.hadoop.util.ToolRunner;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	import java.io.IOException;
   	import java.util.Iterator;
   	import java.util.StringTokenizer;
   	/**
   	* Created by gpzpati on 4/17/14.
   	*/
   	public class S3MapReduce extends Configured implements Tool {
   	Logger logger = LoggerFactory.getLogger(S3Mapper.class);
   	public static class S3Mapper extends Mapper<LongWritable, Text, Text,IntWritable>{
   	Logger logger = LoggerFactory.getLogger(S3Mapper.class);
   	private final static IntWritable one = new IntWritable(1);
   	private Text word = new Text();
   	@Override
   	protected void map(LongWritable key, Text value,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering S3Mapper.map() " + this);
   	String line = value.toString();
   	StringTokenizer st = new StringTokenizer(line," ");
   	while(st.hasMoreTokens()){
   	word.set(st.nextToken());
   	context.write(word,one);
   	}
   	logger.debug("Exiting S3Mapper.map()");
   	}
   	}
   	public static class S3Reducer extends Reducer<Text, IntWritable, Text, IntWritable>{
   	Logger logger = LoggerFactory.getLogger(S3Reducer.class);
   	@Override
   	protected void reduce(Text key, Iterable<IntWritable> values,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering S3Reducer.reduce() " + this);
   	int sum = 0;
   	Iterator<IntWritable> valuesIt = values.iterator();
   	while(valuesIt.hasNext()){
   	sum = sum + valuesIt.next().get();
   	}
   	logger.debug(key + " -> " + sum);
   	context.write(key, new IntWritable(sum));
   	logger.debug("Exiting S3Reducer.reduce()");
   	}
   	}
   	public static void main(String[] argv)throws Exception{
   	int exitCode = ToolRunner.run(new S3MapReduce(), argv);
   	System.exit(exitCode);
   	}
   	@Override
   	public int run(String[] args) throws Exception {
   	if (args.length != 2) {
   	System.err.printf("Usage: %s [generic options] <input> <output>\n",
   	getClass().getSimpleName());
   	ToolRunner.printGenericCommandUsage(System.err);
   	return -1;
   	}
   	Job job = new org.apache.hadoop.mapreduce.Job();
   	job.setJarByClass(S3MapReduce.class);
   	job.setJobName("S3MapReduce");
   	job.getConfiguration().set("fs.s3n.awsAccessKeyId", "awsAccessKey");
   	job.getConfiguration().set("fs.s3n.awsSecretAccessKey","awsSecretAccessKey");
   	job.getConfiguration().set("fs.default.name","s3n://com.spnotes.hadoop.input.books");
   	System.out.println("Input path " + args[0]);
   	System.out.println("Oupput path " + args[1]);
   	FileInputFormat.addInputPath(job, new Path(args[0]));
   	FileOutputFormat.setOutputPath(job, new Path(args[1]));
   	job.setMapperClass(S3Mapper.class);
   	job.setReducerClass(S3Reducer.class);
   	job.setOutputKeyClass(Text.class);
   	job.setOutputValueClass(IntWritable.class);
   	job.setOutputFormatClass(MapFileOutputFormat.class);
   	int returnValue = job.waitForCompletion(true) ? 0:1;
   	System.out.println("job.isSuccessful " + job.isSuccessful());
   	return returnValue;
   	}
   	}

   view raw S3MapReduce.java hosted with ❤ by GitHub
4. Now last step is to execute this program, it takes 2 inputs, You can just right click on your S3MapReduce program and say execute with following 2 parameters

   
   s3n://com.spnotes.hadoop.wordcount.books s3n://com.spnotes.hadoop.wordcount.output/output3
   

5. Once the MapReduce is executed you can check the output by going to S3 console and looking at content of com.spnotes.hadoop.wordcount.output like this
   

Hadoop MapReduce HTTP Notification

Normally MapReduce programs tend to run for long time and you might want to setup a way to get notification when the job is done finishing to find out if the job was executed successfully or not. Hadoop provides a mechanism in which you can get notification on the progress of your MapReduce job. This is two step process

1. First set up job.end.notification.url property with value equal to a web application URL that should get invoked with the progress of job job.getConfiguration().set("job.end.notification.url", "http://localhost:8080/mrnotification/MRNotification/$jobId?status=$jobStatus");
2. Next create a web application that will receive the notification and printout the job status and job name like this

   	package com.spnotes.hadoop;
   	import java.io.IOException;
   	import java.util.Enumeration;
   	import javax.servlet.ServletException;
   	import javax.servlet.http.HttpServlet;
   	import javax.servlet.http.HttpServletRequest;
   	import javax.servlet.http.HttpServletResponse;
   	/**
   	* Servlet implementation class MRNotification
   	*/
   	public class MRNotification extends HttpServlet {
   	private static final long serialVersionUID = 1L;
   	/**
   	* Default constructor.
   	*/
   	public MRNotification() {
   	// TODO Auto-generated constructor stub
   	}
   	/**
   	* @see HttpServlet#doGet(HttpServletRequest request, HttpServletResponse response)
   	*/
   	protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
   	System.out.println("Entering MRNotification.doGet()");
   	process(request,response);
   	System.out.println("Exiting MRNotification.doGet()");
   	}
   	/**
   	* @see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse response)
   	*/
   	protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
   	System.out.println("Entering MRNotification.doPost()");
   	process(request,response);
   	System.out.println("Exiting MRNotification.doPost()");
   	}
   	protected void process(HttpServletRequest request, HttpServletResponse response)throws ServletException, IOException {
   	response.setContentType("text/html");
   	response.getWriter().println("<h1>Hello from MRNotification</h1>");
   	String pathInfo = request.getPathInfo();
   	System.out.println("Job Id -> " +pathInfo.substring(pathInfo.lastIndexOf("/")+1));
   	System.out.println("Job Status -> " + request.getParameter("status"));
   	}
   	}

   view raw MRNotification.java hosted with ❤ by GitHub

Using output of the MapReduce program as input in another MapReduce program - KeyValueTextInputFormat

In the WordCount(HelloWorld) MapReduce program i blogged about how to create a MapReduce program that takes a text file as input and generates output which tells you frequency of each word in the input file. I wanted to take that a step further by reading the output generated by the first MapReduce and figure out which word is used most frequently and how many times that word is used. So i developed this HadoopWordCountProcessor program to do that.

1. First take a look at the output generated by the HadoopWordCount program, which looks like this. In the HadoopWordCount program i used TextOutputFormat as output format class, this class generates output in which there is one key value pair on every line separated by tab character XXX 3 YYY 3 ZZZ 3 aaa 10 bbb 5 ccc 5 ddd 5 eee 5 fff 5 ggg 5 hhh 5 iii 5
2. First create a WordCountProcessorMapper.java program like this, this class receives Text class as Key and value, Only thing i am doing here is converting the Text key into IntWritable and then writing it into output.

   	package com.spnotes.hadoop;
   	import java.io.IOException;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Mapper;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	public class WordCountProcessorMapper extends Mapper<Text, Text, Text, IntWritable>{
   	Logger logger = LoggerFactory.getLogger(WordCountProcessorMapper.class);
   	private final static IntWritable valueInt = new IntWritable(1);
   	@Override
   	protected void map(Text key, Text value,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering WordCountProcessorMapper.map()");
   	valueInt.set(Integer.parseInt(value.toString()));
   	context.write(key, valueInt);
   	logger.debug("Exiting WordCountProcessorMapper.map()");
   	}
   	}

   view raw WordCountProcessorMapper.java hosted with ❤ by GitHub
3. The reducer class is the place where i am getting all the words as key and their frequency as value. In this class i am keeping track of highest frequency word (You will have to copy the key and value of highest occuring word into a local variable for it to work because hadoop reuses key and values object sent to reducer)

   	package com.spnotes.hadoop;
   	import java.io.IOException;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Reducer;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	public class WordCountProcessorReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
   	Logger logger = LoggerFactory.getLogger(WordCountProcessorReducer.class);
   	private Text highestKey;
   	private IntWritable highestCount;
   	@Override
   	protected void reduce(Text key, Iterable<IntWritable> list,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering WordCount2Reducer.reduce()");
   	IntWritable value = list.iterator().next();
   	if(highestCount == null){
   	highestCount = new IntWritable(value.get());
   	highestKey = new Text(key.toString());
   	}else if(value.compareTo(highestCount) > 0){
   	highestCount = new IntWritable(value.get());
   	highestKey = new Text(key.toString());
   	}
   	logger.debug("Exiting WordCount2Reducer.reduce()");
   	}
   	@Override
   	protected void cleanup(org.apache.hadoop.mapreduce.Reducer.Context context)
   	throws IOException, InterruptedException {
   	context.write(highestKey, new Text(highestCount+""));
   	}
   	}

   view raw WordCountProcessorReducer.java hosted with ❤ by GitHub
4. The last step is to create a Driver class, note one thing about the Driver class, i am setting job.setInputFormatClass(KeyValueTextInputFormat.class);, in this i am setting KeyValueTextInputFormat as input format class. Once i do that hadoop takes care of reading the input and breaking it into key and value and passing to my Mapper class

   	package com.spnotes.hadoop;
   	import org.apache.hadoop.conf.Configured;
   	import org.apache.hadoop.fs.Path;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Job;
   	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   	import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
   	import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
   	import org.apache.hadoop.util.Tool;
   	import org.apache.hadoop.util.ToolRunner;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	public class WordCountProcessor extends Configured implements Tool {
   	Logger logger = LoggerFactory.getLogger(WordCountProcessor.class);
   	public static void main(String[] args)throws Exception {
   	int exitCode = ToolRunner.run(new WordCountProcessor(), args);
   	System.exit(exitCode);
   	}
   	@Override
   	public int run(String[] args) throws Exception {
   	if (args.length != 2) {
   	System.err.printf("Usage: %s [generic options] <input> <output>\n",
   	getClass().getSimpleName());
   	ToolRunner.printGenericCommandUsage(System.err);
   	return -1;
   	}
   	Job job = new org.apache.hadoop.mapreduce.Job();
   	job.setJarByClass(WordCountProcessor.class);
   	job.setJobName("WordCounter2");
   	logger.info("Input path " + args[0]);
   	logger.info("Oupput path " + args[1]);
   	FileInputFormat.addInputPath(job, new Path(args[0]));
   	FileOutputFormat.setOutputPath(job, new Path(args[1]));
   	job.setInputFormatClass(KeyValueTextInputFormat.class);
   	job.setMapperClass(WordCountProcessorMapper.class);
   	job.setMapOutputValueClass(IntWritable.class);
   	job.setReducerClass(WordCountProcessorReducer.class);
   	job.setOutputKeyClass(Text.class);
   	job.setOutputValueClass(Text.class);
   	int returnValue = job.waitForCompletion(true) ? 0:1;
   	System.out.println("job.isSuccessful " + job.isSuccessful());
   	return returnValue;
   	}
   	}

   view raw WordCountProcessor.java hosted with ❤ by GitHub
5. Next step is to execute the WordCountProcessor.java class with the output of the first MapReduce program as input by passing couple of arguments like this file:////Users/gpzpati/hadoop/output/wordcount file:///Users/gpzpati/hadoop/output/wordcount2 It will generate output like this. Which says aaa is the most frequently used word and it appeared 10 times aaa 10

Use MRUnit for testing your MapReduce program

In the WordCount(HelloWorld) MapReduce program entry i blogged about how you can create your WordCount (HelloWorld) MapReduce program. You can use Apache MRUnit which is a unit testing framework for testing your MapReduce program. I used MRUnit for developing unit tests for my WordCount program.

1. First i did create a unit test for my WordCountMapper class like this, Basic idea here is you set input and expected output on the test class and then execute the test by calling mapDriver.runTest()

   	package com.spnotes.hadoop;
   	import static org.junit.Assert.*;
   	import java.util.ArrayList;
   	import java.util.List;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.LongWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mrunit.mapreduce.MapDriver;
   	import org.apache.hadoop.mrunit.types.Pair;
   	import org.junit.Before;
   	import org.junit.Test;
   	public class WordCountMapperTest {
   	MapDriver<LongWritable, Text, Text, IntWritable> mapDriver;
   	@Before
   	public void setup(){
   	WordCountMapper mapper = new WordCountMapper();
   	mapDriver = MapDriver.newMapDriver(mapper);// <LongWritable, Text, Text, IntWritable>.
   	}
   	@Test
   	public void testOneWord() throws Exception{
   	mapDriver.withInput(new LongWritable(1), new Text("This"));
   	mapDriver.withOutput(new Text("This"), new IntWritable(1));
   	mapDriver.runTest();
   	}
   	@Test
   	public void testThreeWords() throws Exception{
   	mapDriver.withInput(new LongWritable(1), new Text("This is test"));
   	List<Pair<Text, IntWritable>> output = new ArrayList<Pair<Text,IntWritable>>();
   	output.add(new Pair<Text, IntWritable>(new Text("This"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("is"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("test"), new IntWritable(1)));
   	mapDriver.withAllOutput(output);
   	mapDriver.runTest();
   	}
   	@Test
   	public void testRepeatedWord() throws Exception{
   	mapDriver.withInput(new LongWritable(1), new Text("test test test"));
   	List<Pair<Text, IntWritable>> output = new ArrayList<Pair<Text,IntWritable>>();
   	output.add(new Pair<Text, IntWritable>(new Text("test"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("test"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("test"), new IntWritable(1)));
   	mapDriver.withAllOutput(output);
   	mapDriver.runTest();
   	}
   	@Test
   	public void testOrderingOfKeys() throws Exception{
   	mapDriver.withInput(new LongWritable(1), new Text("ccc bbb aaa"));
   	List<Pair<Text, IntWritable>> output = new ArrayList<Pair<Text,IntWritable>>();
   	output.add(new Pair<Text, IntWritable>(new Text("ccc"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("bbb"), new IntWritable(1)));
   	output.add(new Pair<Text, IntWritable>(new Text("aaa"), new IntWritable(1)));
   	mapDriver.withAllOutput(output);
   	mapDriver.runTest();
   	}
   	}

   view raw WordCountMapperTest.java hosted with ❤ by GitHub
2. Then i did create a unit test for my WordCountReducer class like this

   	package com.spnotes.hadoop;
   	import static org.junit.Assert.*;
   	import java.util.ArrayList;
   	import java.util.List;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
   	import org.junit.Before;
   	import org.junit.Test;
   	public class WordCountReducerTest {
   	private ReduceDriver<Text, IntWritable, Text, IntWritable> reduceDriver;
   	@Before
   	public void setup(){
   	WordCountReducer wordCountReducer = new WordCountReducer();
   	reduceDriver = ReduceDriver.newReduceDriver(new WordCountReducer());
   	}
   	@Test
   	public void testSimpleReduce(){
   	List<IntWritable> wordCountList = new ArrayList<IntWritable>();
   	wordCountList.add(new IntWritable(1));
   	wordCountList.add(new IntWritable(1));
   	wordCountList.add(new IntWritable(1));
   	reduceDriver.withInput(new Text("test"), wordCountList);
   	reduceDriver.withOutput(new Text("test"), new IntWritable(3));
   	}
   	}

   view raw WordCountReducerTest.java hosted with ❤ by GitHub
3. Last part was to develop a end to end test, in this case you setup both mapper and reducer class that you want to set and then run it end to end

   	package com.spnotes.hadoop;
   	import static org.junit.Assert.*;
   	import java.util.ArrayList;
   	import java.util.List;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.LongWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mrunit.mapreduce.MapDriver;
   	import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
   	import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
   	import org.apache.hadoop.mrunit.types.Pair;
   	import org.junit.Before;
   	import org.junit.Test;
   	public class WordCountMRTest {
   	MapReduceDriver<LongWritable, Text, Text, IntWritable, Text, IntWritable> mapReduceDriver;
   	MapDriver<LongWritable, Text, Text, IntWritable> mapDriver;
   	ReduceDriver<Text, IntWritable, Text, IntWritable> reduceDriver;
   	@Before
   	public void setUp() {
   	WordCountMapper mapper = new WordCountMapper();
   	WordCountReducer reducer = new WordCountReducer();
   	mapDriver = new MapDriver<LongWritable, Text, Text, IntWritable>();
   	mapDriver.setMapper(mapper);
   	reduceDriver = new ReduceDriver<Text, IntWritable, Text, IntWritable>();
   	reduceDriver.setReducer(reducer);
   	mapReduceDriver = new MapReduceDriver<LongWritable, Text, Text, IntWritable, Text, IntWritable>();
   	mapReduceDriver.setMapper(mapper);
   	mapReduceDriver.setReducer(reducer);
   	}
   	@Test
   	public void testSimple()throws Exception {
   	mapReduceDriver.withInput(new LongWritable(1), new Text("This test is simple test"));
   	List<Pair<Text, IntWritable>> outputs = new ArrayList<Pair<Text,IntWritable>>();
   	outputs.add(new Pair<Text, IntWritable>(new Text("This"),new IntWritable(1)));
   	outputs.add(new Pair<Text, IntWritable>(new Text("is"),new IntWritable(1)));
   	outputs.add(new Pair<Text, IntWritable>(new Text("simple"),new IntWritable(1)));
   	outputs.add(new Pair<Text, IntWritable>(new Text("test"),new IntWritable(2)));
   	mapReduceDriver.withAllOutput(outputs);
   	mapReduceDriver.runTest();
   	}
   	}

   view raw WordCountMRTest.java hosted with ❤ by GitHub

Maven script for running Hadoop programs

In the WordCount(HelloWorld) MapReduce program blog i talked about how to create a WordCount MapReduce program. While developing MapReduce program, i follow this pattern in which first i develop it using MRUnit test driven development, then i execute it on local using driver. But last step for me is always to copy this program to my Cloudera VM and execute it. I build this maven script to help me with the last part, which is to scp the deployment .jar file to the Cloudera Hadoop VM and then execute it. This is the script i use When i create a new MapReduce program, i have to make couple of changes in it but i can always reuse most of it

1. Change the value of scp.host to point to my hadoop vm, if you changed the username and password on your VM you will have to change it too
2. Next i have to change the value of mainClass attribute to point to correct class for the MapReduce program that i am developing. In this case name of the driver class its com.spnotes.hadoop.WordCountDriver
3. Then i have to change the value of command attribute in sshexec element. THe command is made up of different parts

   hadoop jar ${scp.dirCopyTo}/${project.build.finalName}.jar books/dickens.txt wordcount/outputs

   in this ${scp.dirCopyTo}/${project.build.finalName}.jar points to the .jar file that is being scp to the VM. books/dickens.txt is path of the input text file, in this case i am using hdfs as input location which points to hdfs://localhost/user/cloudera/books/dickens.txt and the output of mapreduce will get generated in hdfs://localhost/user/cloudera/wordcount/outputs

You can run maven antrun:run command for executing the maven script task that deploys the maperduce jar to the cloudera vm and executes it. You can execute the full project from here

	<?xml version="1.0" encoding="UTF-8"?>
	<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
	<groupId>com.spnotes.hadoop</groupId>
	<artifactId>HadoopWordCount</artifactId>
	<version>1.0-SNAPSHOT</version>
	<properties>
	<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
	<hadoop.version>2.0.0-cdh4.0.0</hadoop.version>
	<scp.user>cloudera</scp.user>
	<scp.password>cloudera</scp.password>
	<scp.host>172.16.225.176</scp.host>
	<scp.dirCopyTo>/home/cloudera/test</scp.dirCopyTo>
	</properties>
	<dependencies>
	<dependency>
	<groupId>org.apache.hadoop</groupId>
	<artifactId>hadoop-hdfs</artifactId>
	<version>${hadoop.version}</version>
	</dependency>
	<dependency>
	<groupId>org.apache.hadoop</groupId>
	<artifactId>hadoop-auth</artifactId>
	<version>${hadoop.version}</version>
	</dependency>
	<dependency>
	<groupId>org.apache.hadoop</groupId>
	<artifactId>hadoop-common</artifactId>
	<version>${hadoop.version}</version>
	</dependency>
	<dependency>
	<groupId>org.apache.hadoop</groupId>
	<artifactId>hadoop-core</artifactId>
	<version>2.0.0-mr1-cdh4.0.1</version>
	</dependency>
	<dependency>
	<groupId>junit</groupId>
	<artifactId>junit</artifactId>
	<version>4.10</version>
	<scope>test</scope>
	</dependency>
	<dependency>
	<groupId>org.apache.mrunit</groupId>
	<artifactId>mrunit</artifactId>
	<version>1.0.0</version>
	<classifier>hadoop2</classifier>
	<scope>test</scope>
	</dependency>
	</dependencies>
	<build>
	<finalName>Sorting</finalName>
	<plugins>
	<plugin>
	<groupId>org.apache.maven.plugins</groupId>
	<artifactId>maven-compiler-plugin</artifactId>
	<version>2.1</version>
	<configuration>
	<source>1.6</source>
	<target>1.6</target>
	</configuration>
	</plugin>
	<plugin>
	<groupId>org.apache.maven.plugins</groupId>
	<artifactId>maven-jar-plugin</artifactId>
	<configuration>
	<archive>
	<manifest>
	<mainClass>com.spnotes.hadoop.WordCountDriver</mainClass>
	</manifest>
	</archive>
	</configuration>
	</plugin>
	<plugin>
	<artifactId>maven-antrun-plugin</artifactId>
	<configuration>
	<tasks>
	<scp todir="${scp.user}:${scp.password}@${scp.host}:/${scp.dirCopyTo}"
	file="${project.build.directory}/${project.build.finalName}.jar"
	trust="true" failonerror="false">
	</scp>
	<sshexec host="${scp.host}" username="${scp.user}"
	password="${scp.password}"
	command="hadoop jar ${scp.dirCopyTo}/${project.build.finalName}.jar books/dickens.txt wordcount/outputs2" />
	</tasks>
	</configuration>
	<dependencies>
	<dependency>
	<groupId>ant</groupId>
	<artifactId>ant-jsch</artifactId>
	<version>1.6.5</version>
	</dependency>
	<dependency>
	<groupId>com.jcraft</groupId>
	<artifactId>jsch</artifactId>
	<version>0.1.42</version>
	</dependency>
	</dependencies>
	</plugin>
	</plugins>
	</build>
	<repositories>
	<repository>
	<id>Apache Repository</id>
	<url>https://repository.apache.org/content/repositories</url>
	</repository>
	<repository>
	<id>cloudera</id>
	<url>https://repository.cloudera.com/artifactory/cloudera-repos</url>
	</repository>
	</repositories>
	</project>

view raw pom.xml hosted with ❤ by GitHub

WordCount(HelloWorld) MapReduce program

I am learning about MapReduce and in order to experiment with MapReduce, i created this simple program which takes a text file as input and then generate a output that prints how frequently a word appeared in the text file. You can download the source code for the program from here

1. First i started by creating a simple Mapper which receives the content of the text file one line at a time, the Mapper takes care of splitting the content into words and then it writes every word into output and sets frequency count for that word to 1, by calling context.write(word,one). In this case the word becomes key and count becomes value

   	package com.spnotes.hadoop;
   	import java.io.IOException;
   	import java.util.StringTokenizer;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.LongWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Mapper;
   	import org.slf4j.LoggerFactory;
   	public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
   	org.slf4j.Logger logger = LoggerFactory.getLogger(WordCountMapper.class);
   	private final static IntWritable one = new IntWritable(1);
   	private Text word = new Text();
   	@Override
   	protected void map(LongWritable key, Text value,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering WordCountMapper.map() " + this);
   	String line = value.toString();
   	StringTokenizer st = new StringTokenizer(line," ");
   	while(st.hasMoreTokens()){
   	word.set(st.nextToken());
   	context.write(word,one);
   	}
   	logger.debug("Exiting WordCountMapper.map()");
   	}
   	}

   view raw WordCountMapper.java hosted with ❤ by GitHub
2. Next i had to develop a Reducer class which, receives word as key and value is list of all the counts for example if your input file is simple text like aaa bbb ccc aaa, then reduce class will get called with aaa - [1, 1], bbb -[1] and ccc - [1] as input. Hadoop framework takes care of collecting output of Mapper and then converting it into key -[value,value] format. In the reducer only thing that i had to do was to iterate through all the values and come up with a count. Once i have that i write it as output of Reducer by calling context.write(key, new IntWritable(sum));

   	package com.spnotes.hadoop;
   	import java.io.IOException;
   	import java.util.Iterator;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Reducer;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
   	Logger logger = LoggerFactory.getLogger(WordCountReducer.class);
   	@Override
   	protected void reduce(Text key, Iterable<IntWritable> values,
   	Context context)
   	throws IOException, InterruptedException {
   	logger.debug("Entering WordCountReducer.reduce() " + this);
   	int sum = 0;
   	Iterator<IntWritable> valuesIt = values.iterator();
   	while(valuesIt.hasNext()){
   	sum = sum + valuesIt.next().get();
   	}
   	logger.debug(key + " -> " + sum);
   	context.write(key, new IntWritable(sum));
   	logger.debug("Exiting WordCountReducer.reduce()");
   	}
   	}

   view raw WordCountReducer.java hosted with ❤ by GitHub
3. The last part is creating WordCountDriver.java, which is a Java program that sets up Hadoop Framework by setting up inputs, defining outputs and also specifying name of the Mapper and Reducer class. After initializing Hadoop it calls job.waitForCompletion(true), this method will take care of passing the control to Hadoop framework and wait for the job to complete

   	package com.spnotes.hadoop;
   	import org.apache.hadoop.conf.Configured;
   	import org.apache.hadoop.fs.Path;
   	import org.apache.hadoop.io.IntWritable;
   	import org.apache.hadoop.io.LongWritable;
   	import org.apache.hadoop.io.Text;
   	import org.apache.hadoop.mapreduce.Job;
   	import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   	import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
   	import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
   	import org.apache.hadoop.util.Tool;
   	import org.apache.hadoop.util.ToolRunner;
   	import org.slf4j.Logger;
   	import org.slf4j.LoggerFactory;
   	public class WordCountDriver extends Configured implements Tool{
   	Logger logger = LoggerFactory.getLogger(WordCountDriver.class);
   	public static void main(String[] args) throws Exception{
   	int exitCode = ToolRunner.run(new WordCountDriver(), args);
   	System.exit(exitCode);
   	}
   	public int run(String[] args) throws Exception {
   	if (args.length != 2) {
   	System.err.printf("Usage: %s [generic options] <input> <output>\n",
   	getClass().getSimpleName());
   	ToolRunner.printGenericCommandUsage(System.err);
   	return -1;
   	}
   	Job job = new org.apache.hadoop.mapreduce.Job();
   	job.setJarByClass(WordCountDriver.class);
   	job.setJobName("WordCounter");
   	logger.info("Input path " + args[0]);
   	logger.info("Oupput path " + args[1]);
   	FileInputFormat.addInputPath(job, new Path(args[0]));
   	FileOutputFormat.setOutputPath(job, new Path(args[1]));
   	job.setOutputKeyClass(Text.class);
   	job.setOutputValueClass(IntWritable.class);
   	job.setOutputFormatClass(TextOutputFormat.class);
   	job.setMapperClass(WordCountMapper.class);
   	job.setReducerClass(WordCountReducer.class);
   	int returnValue = job.waitForCompletion(true) ? 0:1;
   	System.out.println("job.isSuccessful " + job.isSuccessful());
   	return returnValue;
   	}
   	}

   view raw WordCountDriver.java hosted with ❤ by GitHub
4. Now you can either use one of the existing .txt file on your machine or you can create a text file like this

   
   hhh eee  iii bbb ccc fff ddd  ggg aaa aaa
   XXX YYY ZZZ
   hhh eee  iii bbb ccc fff ddd  ggg aaa aaa 
   XXX YYY ZZZ
   hhh eee  iii bbb ccc fff ddd  ggg aaa aaa 
   XXX YYY ZZZ
   hhh eee  iii bbb ccc fff ddd  ggg aaa aaa 
   hhh eee  iii bbb ccc fff ddd  ggg aaa aaa
   

5. Last step is to run your Hadoop program, if you used the Eclipse or some other IDE for developing your code, you can run your program directly by running WordCountDriver.java directly. This program takes 2 parameters, in my case since the input file is on local file system and i want the output to get stored on local file system too, i pass following 2 parameters

   
   file:///Users/sunil/hadoop/sorttest.txt file:///Users/sunil/hadoop/output/wordcount
   

6. Once the program is finished successfully, you would be able to see part-r-00000 file created on your local machine at /Users/sunil/hadoop/output/wordcount, if you open it you should see output like this

   
   XXX     3
   YYY     3
   ZZZ     3
   aaa     10
   bbb     5
   ccc     5
   ddd     5
   eee     5
   fff     5
   ggg     5
   hhh     5
   iii     5
   

If you want to run this program with bigger text file then you can download few good classical books from Algorithm site data section