Programing Mapreduce
(Hadoop)
with Eclipse
1. Prepare:
● System:
○ Ubuntu 12.04
○ Hadoop 1.0.4
● Requirement:
Eclipse IDE for Java EE Developers 1.7.0
Install eclipse IDE in Ubuntu using Ubuntu software center
Install java 7 and plugins.
● Append two codes to /etc/[Link] to setup java Class path
export JAVA_Home/usr/java/jdk1.6.0_25
export PATH=$PATH:$JAVA_HOME/bin
export HADOOP_HOME/home/hadoop/hadoop-1.0.4
export HADOOP_LIB=$HADOOP_HOME/lib
Step 1. Install and start the Hadoop server
In this section:I assume your Hadoop installation is [Link] single Node setup
Start Hadoop:
hadoop@slavenode1:~/apache/hadoop-1.0.4$ bin/[Link]
hadoop@slavenode1:~/apache/hadoop-1.0.4$ jps
6098 JobTracker
8024 Jps
5783 DataNode
5997 SecondaryNameNode
5571 NameNode
6310 TaskTracker
(Make sure NameNode, DataNode, JobTracker, TaskTracker, SecondaryNameNode are running)
● Start with Eclipse :
○ Click on the eclipse icon
○ ○ create a workspace where you need to store your project.
○ click ok.
● Then There is a screen appears like this.
○ Then Right click on your Project Explore space
○ click on Project
○ click on java project and Next.
○ Next
○ Put your project name and Finish.
○ Right click on your project -->Build Path -->Configure Build Path-->click
○ Then add External jar-->select jars-->ok
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ [Link]
○ Project configuration is [Link] your Project name is SimpleWordCount.....
○ Now expand your project and you’ll get src folder.
○ Delete the src folder.
○ create a source folder
○ Right click on project-->new-->source Folder
○ click and put src/main/java
○ ok
○ create a package named as [Link] inside src/main/java Folder.
○ now create class named WordCountDriver inside [Link] package.
○ copy the code and paste in [Link]
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class WordCountDriver extends Configured implements Tool{
public static void main(String[] args) throws Exception
{
[Link](new WordCountDriver(),args);
}
@Override
public int run(String[] args) throws Exception {
Job job = new Job(getConf(),"Basic Word Count Job");
[Link]([Link]);
//Map and Reduce
[Link]([Link]);
[Link]([Link]);
[Link]( 1 );
[Link]([Link]);
//the map output
[Link]([Link]);
[Link]([Link]);
//the reduce output
[Link]([Link]);
[Link]([Link]);
[Link](job, new Path(args[0]));
[Link](job, new Path(args[1]));
[Link](true);
return 0;
}
○ Again create a class named WordContMapper inside [Link]
○ Copy the code and paste in [Link]
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
/**
*
* @author training
* Class : WordCountMapper
*
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
/**
* Optimization: Instead of creating the variables in the
*/
@Override
public void map(LongWritable inputKey,Text inputVal,Context context) throws
IOException,InterruptedException
{
String line = [Link]();
String[] splits = [Link]().split("\\W+");
for(String outputKey:splits)
{
[Link](new Text(outputKey), new IntWritable(1));
○ create WordCounterReduce class inside [Link]
○ copy the code and paste in [Link]
import [Link];
import [Link];
import [Link];
import [Link];
public class WordCountReducer extends Reducer<Text,IntWritable,Text, IntWritable>{
@Override
public void reduce(Text key,Iterable<IntWritable> listOfValues,Context context) throws
IOException,InterruptedException
{
int sum=0;
for(IntWritable val:listOfValues)
{
sum = sum + [Link]();
}
[Link](key,new IntWritable(sum));
○ create jar file of the Project .
○ Rigth click on project and take the cursor to Export option
○ click it
○ click on jar file and Next
○ put the jar file name with [Link] click on Finish.
○ Run the jar file in terminal
bin/hadoo jar jarfilename DriverClassname inputfilename outputfilename
○ see the output in Browser
○ localhost:50070/[Link] → enter
● HDFS programming in Java:-
● Below is a code sample of how to read from and write to HDFS in
java.
○ Creating a configuration object: To be able to read from or write to HDFS, you
need to create a Configuration object and pass configuration parameter to it using
hadoop configuration files.
// Conf object will read the HDFS configuration parameters from these XML
// files. You may specify the parameters for your own if you want.
○ If you do not assign the configurations to conf object (using hadoop xml file) your HDFS
operation will be performed on the local file system and not on the HDFS.
● Adding file to HDFS: Create a FileSystem object and use a file stream to add a file.
FileSystem fileSystem = [Link](conf);
// Check if the file already exists
Path path = new Path("/path/to/[Link]");
if ([Link](path)) {
[Link]("File " + dest + " already exists");
return;
}
// Create a new file and write data to it.
FSDataOutputStream out = [Link](path);
InputStream in = new BufferedInputStream(new FileInputStream(
new File(source)));
byte[] b = new byte[1024];
int numBytes = 0;
while ((numBytes = [Link](b)) > 0) {
[Link](b, 0, numBytes);
}
// Close all the file descripters
[Link]();
[Link]();
[Link]();
○ Reading file from HDFS: Create a file stream object to a file in HDFS and read it.
FileSystem fileSystem = [Link](conf);
Path path = new Path("/path/to/[Link]");
if () {
[Link]("File does not exists");
return;
}
FSDataInputStream in = [Link](path);
String filename = [Link]([Link]('/') + 1,
[Link]());
OutputStream out = new BufferedOutputStream(new FileOutputStream(
new File(filename)));
byte[] b = new byte[1024];
int numBytes = 0;
while ((numBytes = [Link](b)) > 0) {
[Link](b, 0, numBytes);
}
[Link]();
[Link]();
[Link]();
3. Deleting file from HDFS: Create a file stream object to a file in HDFS and delete it.
FileSystem fileSystem = [Link](conf);
Path path = new Path("/path/to/[Link]");
if () {
[Link]("File does not exists");
return;
}
// Delete file
[Link](new Path(file), true);
[Link]();
○ Create dir in HDFS: Create a file stream object to a file in HDFS and read it.
FileSystem fileSystem = [Link](conf);
Path path = new Path(dir);
if ([Link](path)) {
[Link]("Dir " + dir + " already not exists");
return;
}
// Create directories
[Link](path);
[Link]();
○ Complete code for Adding ,Reading,Deleting and create directory.
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
import [Link];
public class HDFSClient {
public HDFSClient() {
public void addFile(String source, String dest) throws IOException {
Configuration conf = new Configuration();
// Conf object will read the HDFS configuration parameters from these
// XML files.
[Link](new Path("/opt/hadoop-0.20.0/conf/[Link]"));
[Link](new Path("/opt/hadoop-0.20.0/conf/[Link]"));
FileSystem fileSystem = [Link](conf);
// Get the filename out of the file path
String filename = [Link]([Link]('/') + 1,
[Link]());
// Create the destination path including the filename.
if ([Link]([Link]() - 1) != '/') {
dest = dest + "/" + filename;
} else {
dest = dest + filename;
}
// [Link]("Adding file to " + destination);
// Check if the file already exists
Path path = new Path(dest);
if ([Link](path)) {
[Link]("File " + dest + " already exists");
return;
}
// Create a new file and write data to it.
FSDataOutputStream out = [Link](path);
InputStream in = new BufferedInputStream(new FileInputStream(
new File(source)));
byte[] b = new byte[1024];
int numBytes = 0;
while ((numBytes = [Link](b)) > 0) {
[Link](b, 0, numBytes);
}
// Close all the file descripters
[Link]();
[Link]();
[Link]();
}
public void readFile(String file) throws IOException {
Configuration conf = new Configuration();
[Link](new Path("/opt/hadoop-0.20.0/conf/[Link]"));
FileSystem fileSystem = [Link](conf);
Path path = new Path(file);
if () {
[Link]("File " + file + " does not exists");
return;
}
FSDataInputStream in = [Link](path);
String filename = [Link]([Link]('/') + 1,
[Link]());
OutputStream out = new BufferedOutputStream(new FileOutputStream(
new File(filename)));
byte[] b = new byte[1024];
int numBytes = 0;
while ((numBytes = [Link](b)) > 0) {
[Link](b, 0, numBytes);
}
[Link]();
[Link]();
[Link]();
}
public void deleteFile(String file) throws IOException {
Configuration conf = new Configuration();
[Link](new Path("/opt/hadoop-0.20.0/conf/[Link]"));
FileSystem fileSystem = [Link](conf);
Path path = new Path(file);
if () {
[Link]("File " + file + " does not exists");
return;
}
[Link](new Path(file), true);
[Link]();
}
public void mkdir(String dir) throws IOException {
Configuration conf = new Configuration();
[Link](new Path("/opt/hadoop-0.20.0/conf/[Link]"));
FileSystem fileSystem = [Link](conf);
Path path = new Path(dir);
if ([Link](path)) {
[Link]("Dir " + dir + " already not exists");
return;
}
[Link](path);
[Link]();
}
public static void main(String[] args) throws IOException {
if ([Link] < 1) {
[Link]("Usage: hdfsclient add/read/delete/mkdir" +
" [<local_path> <hdfs_path>]");
[Link](1);
}
HDFSClient client = new HDFSClient();
if (args[0].equals("add")) {
if ([Link] < 3) {
[Link]("Usage: hdfsclient add <local_path> " +
"<hdfs_path>");
[Link](1);
}
[Link](args[1], args[2]);
} else if (args[0].equals("read")) {
if ([Link] < 2) {
[Link]("Usage: hdfsclient read <hdfs_path>");
[Link](1);
}
[Link](args[1]);
} else if (args[0].equals("delete")) {
if ([Link] < 2) {
[Link]("Usage: hdfsclient delete <hdfs_path>");
[Link](1);
}
[Link](args[1]);
} else if (args[0].equals("mkdir")) {
if ([Link] < 2) {
[Link]("Usage: hdfsclient mkdir <hdfs_path>");
[Link](1);
}
[Link](args[1]);
} else {
[Link]("Usage: hdfsclient add/read/delete/mkdir" +
" [<local_path> <hdfs_path>]");
[Link](1);
}
[Link]("Done!");
}
}
● Debugging & Diagnosys:
○