- Drive Class to run the program
- Mapper Class
- XmlInputFormat class
I am not using reducers to make the example simple. Now Lets do some programming to work out these things.
Driver Class:
Here is the code for driver class. which is explained below.
The code is very easy. In start tag and end tag you need to define one record
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
*
* @author root
*/
public class XmlReader {
/**
*
* @author kiran Bhakre
*/
public static void main(String[] args) {
try {
runJob(args[0], args[1]);
} catch (IOException ex) {
//Logger.getLogger(XmlReader.class.getName()).log(Level.SEVERE, null, ex);
}
}
public static void runJob(String input,
String output ) throws IOException {
Configuration conf = new Configuration();
conf.set("xmlinput.start","<property>");
conf.set("xmlinput.end", "</property>");
conf
.set(
"io.serializations",
"org.apache.hadoop.io.serializer.JavaSerialization,org.apache.hadoop.io.serializer.WritableSerialization");
Job job = new Job(conf, "jobName");
FileInputFormat.setInputPaths(job, input);
job.setJarByClass(XmlReader.class);
job.setMapperClass(XmlReadMapper.class);
job.setNumReduceTasks(0);
job.setInputFormatClass(XmlInputFormat.class);
job.setOutputKeyClass(NullWritable.class);
job.setOutputValueClass(Text.class);
Path outPath = new Path(output);
FileOutputFormat.setOutputPath(job, outPath);
FileSystem dfs = FileSystem.get(outPath.toUri(), conf);
if (dfs.exists(outPath)) {
dfs.delete(outPath, true);
}
try {
job.waitForCompletion(true);
} catch (InterruptedException ex) {
//Logger.getLogger(XmlReader.class.getName()).log(Level.SEVERE, null, ex);
} catch (ClassNotFoundException ex) {
//Logger.getLogger(XmlReader.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
conf.set("xmlinput.start", "<startingTag>");
conf.set("xmlinput.end", "</endingTag>");
Then you need to set input path, output path which i am taking as command line arguments, need to set mapper class.
Next we will define our mapper.
Mapper:
To parse the xml files, you need some parser library, There are many ways to parse xml file in java like using SAX, DOM parser. I have used jdom library to parse the xml file. Here is the code for mapper class which is explained below.
import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.jdom.Document;
import org.jdom.Element;
import org.jdom.JDOMException;
import org.jdom.input.SAXBuilder;
/**
*
* @author kiran Bhakre
*/
public class XmlReadMapper extends
Mapper<LongWritable, Text, NullWritable, Text> {
@Override
public void map(LongWritable key, Text value1,Context context) throws IOException, InterruptedException {
String xmlString = value1.toString();
SAXBuilder builder = new SAXBuilder();
Reader in = new StringReader(xmlString);
String value="";
try {
Document doc = builder.build(in);
Element root = doc.getRootElement();
String tag1 =root.getChild("tag").getChild("tag1").getTextTrim() ;
String tag2 =root.getChild("tag").getChild("tag1").getChild("tag2").getTextTrim();
value= tag1+ ","+tag2;
context.write(NullWritable.get(), new Text(value));
} catch (JDOMException ex) {
// Logger.getLogger(XmlReadMapper.class.getName()).log(Level.SEVERE, null, ex);
} catch (IOException ex) {
// Logger.getLogger(XmlReadMapper.class.getName()).log(Level.SEVERE, null, ex);
}
}
}
The code is very simple, you are getting the record in value1 and then parsing the data and then sending the data using
context.write(NullWritable.get(), new Text(value));
I did not require key so i use NullWritable and value contains comma delimited record after parsing.
Next, i am also providing the Mahout XMLInputFormat class code which is compatible with new Hadoop API.
Mahout XMLinputFormat (Compatible with New Hadoop API):
import java.io.IOException;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
/**
* @author kiran Bhakre
* Reads element in records that are delimited by a specific Start and end tag.
*/
public class XmlInputFormat extends TextInputFormat {
public static final String START_TAG_KEY = "xmlinput.start";
public static final String END_TAG_KEY = "xmlinput.end";
@Override
public RecordReader<LongWritable,Text> createRecordReader(InputSplit is, TaskAttemptContext tac) {
return new XmlRecordReader();
}
public static class XmlRecordReader extends RecordReader<LongWritable,Text> {
private byte[] startTag;
private byte[] endTag;
private long start;
private long end;
private FSDataInputStream fsdatainstream;
private DataOutputBuffer buffer = new DataOutputBuffer();
private LongWritable key = new LongWritable();
private Text value = new Text();
@Override
public void initialize(InputSplit is, TaskAttemptContext tac) throws IOException, InterruptedException {
FileSplit fileSplit= (FileSplit) is;
startTag = tac.getConfiguration().get(START_TAG_KEY).getBytes("utf-8");
endTag = tac.getConfiguration().get(END_TAG_KEY).getBytes("utf-8");
start = fileSplit.getStart();
end = start + fileSplit.getLength();
Path file = fileSplit.getPath();
FileSystem fs = file.getFileSystem(tac.getConfiguration());
fsdatainstream = fs.open(fileSplit.getPath());
fsdatainstream.seek(start);
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
if (fsdatainstream.getPos() < end) {
if (readUntilMatch(startTag, false)) {
try {
buffer.write(startTag);
if (readUntilMatch(endTag, true)) {
value.set(buffer.getData(), 0, buffer.getLength());
key.set(fsdatainstream.getPos());
return true;
}
} finally {
buffer.reset();
}
}
}
return false;
}
@Override
public LongWritable getCurrentKey() throws IOException, InterruptedException {
return key;
}
@Override
public Text getCurrentValue() throws IOException, InterruptedException {
return value;
}
@Override
public float getProgress() throws IOException, InterruptedException {
return (fsdatainstream.getPos() - start) / (float) (end - start);
}
@Override
public void close() throws IOException {
fsdatainstream.close();
}
private boolean readUntilMatch(byte[] match, boolean withinBlock) throws IOException {
int i = 0;
while (true) {
int b = fsdatainstream.read();
// end of file:
if (b == -1) return false;
// save to buffer:
if (withinBlock) buffer.write(b);
// check if we're matching:
if (b == match[i]) {
i++;
if (i >= match.length) return true;
} else i = 0;
// see if we've passed the stop point:
if (!withinBlock && i == 0 && fsdatainstream.getPos() >= end) return false;
}
}
}
}
To run this code, include the necessary jar files (jdom.jar,hadoop-core.jar) and you also need to make a single jar file. You can find the instructions to make a single jar file on the following link
http://java.sun.com/developer/technicalArticles/java_warehouse/single_jar/
Next, give the following command on the terminal to run the job.
hadoop jar xmlreader.jar /user/root/Data/file.xml output/
Conclusion:
In this way, we can process large amount of xml files using hadoop and Mahout XML input format.
