MapReduce code to retrieve employee fields from employee file

I have created a file emp. Lets try a simple mapreduce program to retrieve the fields from employee file. Below is the file and the sample records with headers.

emp
empid,empname,deptid,salary
1,swetha,1,50000
2,Nikhil,2,75000
3,Varun,3,75000

First move the file to HDFS.
$ hadoop dfs -put emp emp

$hadoop dfs -ls
Found 1 item
-rw-r--r--   1 swetha supergroup         81 2015-01-30 17:31 /user/swetha/emp

Lets try to select empname and salary fields from emp table. Below is the mapreduce for the same.

The below code is synonymous to the below SQL:

select empname,salary from emp;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import java.io.*;


//Displays EmpName and EmpID
public class empDetails extends Configured implements Tool {

    static class empMap extends Mapper<LongWritable, Text, Text, Text> {

        @Override
        public void map( LongWritable key,  Text value, Context context)
                throws IOException,InterruptedException{
        String a = value.toString();
        String[] b = a.split(",");
        String empName = b[1];
        String salary = b[3];
        context.write(new Text(empName), new Text(salary));
        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Job job = new Job();
        job.setJarByClass(getClass());
        job.setJobName("Employee Details");
       
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
       
        job.setMapperClass(empMap.class);
        job.setNumReduceTasks(0);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
       
        return job.waitForCompletion(true)?0 : 1;
    }

    public static void main(String[] args) throws Exception{
        int exitcode = ToolRunner.run(new empDetails() , args);
        System.exit(exitcode);
    }
}
Pack the file into a empDetails.jar file and run it on hadoop environment.

$ hadoop jar empDetails.jar empDetails emp out1
15/01/30 19:26:45 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
15/01/30 19:26:46 INFO input.FileInputFormat: Total input paths to process : 1
15/01/30 19:26:46 INFO util.NativeCodeLoader: Loaded the native-hadoop library
15/01/30 19:26:46 WARN snappy.LoadSnappy: Snappy native library not loaded
15/01/30 19:26:46 INFO mapred.JobClient: Running job: job_201501301653_0006
15/01/30 19:26:47 INFO mapred.JobClient:  map 0% reduce 0%
15/01/30 19:26:54 INFO mapred.JobClient:  map 100% reduce 0%
15/01/30 19:26:56 INFO mapred.JobClient: Job complete: job_201501301653_0006
15/01/30 19:26:56 INFO mapred.JobClient: Counters: 19
15/01/30 19:26:56 INFO mapred.JobClient:   Map-Reduce Framework
15/01/30 19:26:56 INFO mapred.JobClient:     Spilled Records=0
15/01/30 19:26:56 INFO mapred.JobClient:     Virtual memory (bytes) snapshot=1845501952
15/01/30 19:26:56 INFO mapred.JobClient:     Map input records=4
15/01/30 19:26:56 INFO mapred.JobClient:     SPLIT_RAW_BYTES=102
15/01/30 19:26:56 INFO mapred.JobClient:     Map output records=4
15/01/30 19:26:56 INFO mapred.JobClient:     Physical memory (bytes) snapshot=75046912
15/01/30 19:26:56 INFO mapred.JobClient:     CPU time spent (ms)=150
15/01/30 19:26:56 INFO mapred.JobClient:     Total committed heap usage (bytes)=32440320
15/01/30 19:26:56 INFO mapred.JobClient:   File Input Format Counters
15/01/30 19:26:56 INFO mapred.JobClient:     Bytes Read=81
15/01/30 19:26:56 INFO mapred.JobClient:   FileSystemCounters
15/01/30 19:26:56 INFO mapred.JobClient:     HDFS_BYTES_READ=183
15/01/30 19:26:56 INFO mapred.JobClient:     FILE_BYTES_WRITTEN=54900
15/01/30 19:26:56 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=55
15/01/30 19:26:56 INFO mapred.JobClient:   Job Counters
15/01/30 19:26:56 INFO mapred.JobClient:     Launched map tasks=1
15/01/30 19:26:56 INFO mapred.JobClient:     SLOTS_MILLIS_REDUCES=0
15/01/30 19:26:56 INFO mapred.JobClient:     Total time spent by all reduces waiting after reserving slots (ms)=0
15/01/30 19:26:56 INFO mapred.JobClient:     SLOTS_MILLIS_MAPS=9046
15/01/30 19:26:56 INFO mapred.JobClient:     Total time spent by all maps waiting after reserving slots (ms)=0
15/01/30 19:26:56 INFO mapred.JobClient:     Data-local map tasks=1
15/01/30 19:26:56 INFO mapred.JobClient:   File Output Format Counters
15/01/30 19:26:56 INFO mapred.JobClient:     Bytes Written=55

Now lets look at the output..

$ hadoop dfs -ls
Found 3 items
-rw-r--r--   1 swetha supergroup         38 2015-01-30 17:31 /user/swetha/dept
-rw-r--r--   1 swetha supergroup         81 2015-01-30 17:31 /user/swetha/emp
drwxr-xr-x   - swetha supergroup          0 2015-01-30 19:26 /user/swetha/out1

since there are no reducers the mapper output is directly written to hdfs.

$ hadoop dfs -lsr out1
-rw-r--r--   1 swetha supergroup          0 2015-01-30 19:26 /user/swetha/out1/_SUCCESS
drwxr-xr-x   - swetha supergroup          0 2015-01-30 19:26 /user/swetha/out1/_logs
drwxr-xr-x   - swetha supergroup          0 2015-01-30 19:26 /user/swetha/out1/_logs/history
-rw-r--r--   1 swetha supergroup       9271 2015-01-30 19:26 /user/swetha/out1/_logs/history/job_201501301653_0006_1422664006709_swetha_Employee+Details
-rw-r--r--   1 swetha supergroup      47336 2015-01-30 19:26 /user/swetha/out1/_logs/history/job_201501301653_0006_conf.xml
-rw-r--r--   1 swetha supergroup         55 2015-01-30 19:26 /user/swetha/out1/part-m-00000

$ hadoop dfs -cat out1/p*
 empname     salary
swetha    50000
Nikhil    75000
Varun    75000


To get the same output without headers, use 

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import java.io.*;


//Displays EmpName and EmpID without Headers
public class empDetailsWithOutHeaders extends Configured implements Tool {

    static class empMap extends Mapper<LongWritable, Text, Text, Text> {

        @Override
        public void map( LongWritable key,  Text value, Context context)
                throws IOException,InterruptedException{
        String a = value.toString();
        String[] b = a.split(",");
        String empName = b[1];
        String salary = b[3];
        if (!(empName.equals("empname")))
            context.write(new Text(empName), new Text(salary));
        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Job job = new Job();
        job.setJarByClass(getClass());
        job.setJobName("Employee Details without Headers");
       
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
       
        job.setMapperClass(empMap.class);
        job.setNumReduceTasks(0);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
       
        return job.waitForCompletion(true)?0 : 1;
    }

    public static void main(String[] args) throws Exception{
        int exitcode = ToolRunner.run(new empDetailsWithOutHeaders() , args);
        System.exit(exitcode);
    }
}

The output will be the following..

 $ hadoop dfs -cat out2/p*
swetha    50000
Nikhil    75000
Varun    75000

VirtualBox/Ubuntu: Full Screen Mode resolution

By default, at startup, ubuntu VMBox presents all the icons in a particular size that is smaller than the actual monitor. But when I switch to either full screen or scaled mode, instead of resizing everything smoothly (like a vector image) it makes everything fuzzy and grainy, and hard to look at.

Eclipse : Error in WordCount MapReduce code : Type java.util.Map$Entry cannot be resolved

I was trying to compile the Hadoop MapReduce WordCount problem and faced the following error. I was using Java 1.8 version. 

Error:

1. The project was not built since its build path is incomplete. Cannot find the class file for java.util.Map$Entry. Fix the build path then try building this project.
2. The type java.util.Map$Entry cannot be resolved. It is indirectly referenced from required .class files.

Resolution:
 Downgrade your Java version from 1.8 to 1.7. Below are the steps I have followed to resolve this error.

1. I have uninstalled Java 1.8 latest version from my computer. Go to Control Panel,  Programs and Features and then uninstall the Java 1.8 Kit.
2. Download and install Java 1.7 version. Change the PATH environment variable on your windows to point to new installed version of Java as in add "C:\Program Files\Java\jdk1.7.0_67\bin" to PATH env. variable.
3. Restart the computer.
4. Open your Eclipse and try to build and compile your project against 1.7. Your error should disappear.

Eclipse : Syntax Error : Parameterized types are only if source level is 1.5

Follow the steps to resolve the above error.

1. Right click on the project and go to project properties.
2. Then go to Java Compiler.
3. Check the Box (Enable project specific settings)
4. Change the compiler compliance level to '1.5'.
5. Also, click the check box for "Use default compliance settings" & click ok.

Rebuild the project. It will be resolved.

Hadoop Integration with Eclipse : Errors while running the WordCount program

Following are the WordCount Mapper,Reducer and Driver classes respectively for the WordCount example picked from YDN.

WordCountMapper.java
--------------------------------------

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.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

public class WordCountMapper extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, IntWritable> {

  private final IntWritable one = new IntWritable(1);
  private Text word = new Text();

  public void map(WritableComparable key, Writable value,
      OutputCollector output, Reporter reporter) throws IOException {

    String line = value.toString();
    StringTokenizer itr = new StringTokenizer(line.toLowerCase());
    while(itr.hasMoreTokens()) {
      word.set(itr.nextToken());
      output.collect(word, one);
    }
  }
}


WordCountReducer.java
-----------------------------------

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

public class WordCountReducer extends MapReduceBase
    implements Reducer<Text, IntWritable, Text, IntWritable> {

  public void reduce(Text key, Iterator values,
      OutputCollector output, Reporter reporter) throws IOException {

    int sum = 0;
    while (values.hasNext()) {
      IntWritable value = (IntWritable) values.next();
      sum += value.get(); // process value
    }

    output.collect(key, new IntWritable(sum));
  }
}

WordCount.java (Driver Class)
--------------------------------------

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;

public class WordCount {

  public static void main(String[] args) {
    JobClient client = new JobClient();
    JobConf conf = new JobConf(WordCount.class);

    // specify output types
    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(IntWritable.class);

    // specify input and output dirs
    FileInputPath.addInputPath(conf, new Path("input"));
    FileOutputPath.addOutputPath(conf, new Path("output"));

    // specify a mapper
    conf.setMapperClass(WordCountMapper.class);

    // specify a reducer
    conf.setReducerClass(WordCountReducer.class);
    conf.setCombinerClass(WordCountReducer.class);

    client.setConf(conf);
    try {
      JobClient.runJob(conf);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

Errors:

1. FileInputPath cannot be resolved
2. FileOutputPath cannot be resolved


Modify
1.  FileInputPath.addInputPath in driver class (WordCount.java) to  FileInputFormat.addInputPath
2.  FileOutputPath.addOutputPath in driver class to FileOutputFormat.setOutputPath

Hope it helps.

ERROR dfs.NameNode: java.io.IOException: Cannot lock storage /tmp/hadoop-hadoop-user/dfs/name. The directory is already locked

The following is the error occurred while formatting the namenode.
----------------------------------------------------------------------------------------------------------------
hadoop-user@hadoop-desk:~/hadoop$ bin/hadoop namenode -format
14/09/14 18:53:27 INFO dfs.NameNode: STARTUP_MSG:
/************************************************************
STARTUP_MSG: Starting NameNode
STARTUP_MSG:   host = hadoop-desk/127.0.1.1
STARTUP_MSG:   args = [-format]
STARTUP_MSG:   version = 0.18.0
STARTUP_MSG:   build = http://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.18 -r 686010; comp                                                                                      iled by 'hadoopqa' on Thu Aug 14 19:48:33 UTC 2008
************************************************************/
Re-format filesystem in /tmp/hadoop-hadoop-user/dfs/name ? (Y or N) Y
14/09/14 18:53:35 INFO fs.FSNamesystem: fsOwner=hadoop-user,hadoop-user,adm,dialout,cdrom,floppy,audio,                                                                                      dip,video,plugdev,fuse,lpadmin,admin
14/09/14 18:53:35 INFO fs.FSNamesystem: supergroup=supergroup
14/09/14 18:53:35 INFO fs.FSNamesystem: isPermissionEnabled=true
14/09/14 18:53:35 INFO dfs.Storage: Cannot lock storage /tmp/hadoop-hadoop-user/dfs/name. The directory                                                                                       is already locked.
14/09/14 18:53:35 ERROR dfs.NameNode: java.io.IOException: Cannot lock storage /tmp/hadoop-hadoop-user/                                                                                      dfs/name. The directory is already locked.
        at org.apache.hadoop.dfs.Storage$StorageDirectory.lock(Storage.java:447)
        at org.apache.hadoop.dfs.FSImage.format(FSImage.java:930)
        at org.apache.hadoop.dfs.FSImage.format(FSImage.java:949)
        at org.apache.hadoop.dfs.NameNode.format(NameNode.java:741)
        at org.apache.hadoop.dfs.NameNode.createNameNode(NameNode.java:822)
        at org.apache.hadoop.dfs.NameNode.main(NameNode.java:839)
14/09/14 18:53:35 INFO dfs.NameNode: SHUTDOWN_MSG:
/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at hadoop-desk/127.0.1.1
************************************************************/
------------------------------------------------------------------------------------------------------

The username with which you are trying to format the name node might not have necessary privileges on the namenode and datanode directories that are listed in the hadoop-site.xml configuration file. Provide the necessary privileges on all directories and sub-directories of name node and data node directories and rerun the command to format.

Getting "Error:Null" while browsing HDFS from mapreduce plugin of Eclipse

While navigating HDFS directory structure in the project explorer, if you are getting Error:NULL check for Advanced Parameters settings. I am using Hadoop 0.18.0 and Eclipse SDK version 3.5.0.

In many of the eclipse versions when you are setting Advanced Parameters, you may not find "hadoop.job.ugi" parameter. And hence missing setting this parameter could result in Error:NULL.

First, set all the required parameters except hadoop.job.ugi (because its missing). Then click Finish.

Now go to your workspace directory and navigate to locations folder.

"\workspace\.metadata\.plugins\org.apache.hadoop.eclipse\locations"

Here open the xml file and add the following.

<property>
< name>hadoop.job.ugi</name>
< value>hadoop-user,ABC</value>
< /property>

Close the eclipse and start it again. Now you should be able to connect to HDFS and view its file system.