Hi
I am trying to convert Mahout xmlInputFormat to new API but this is not
working. The problem which i think is that in old api we have next method
which takes key and value and we can set it in the method
public boolean next(LongWritable key, Text value) throws IOException {
but in new API, the method signature is
public boolean nextKeyValue()
So now i am not sure where i set key and values. Here is the link of xml
Input format with old API
http://grepcode.com/file/repo1.maven.org/maven2/org.apache.mahout/mahout-examples/0.2/org/apache/mahout/classifier/bayes/XmlInputFormat.java
and i am attaching what i have done with it to convert it into new API so
far.
I am not sure about what i have done so far is correct. Kindly help me out
to sort out the problem
Thanks
--
Regards
Shuja-ur-Rehman Baig
http://pk.linkedin.com/in/shujamughal
Cell: +92 3214207445
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.logging.Level;
import java.util.logging.Logger;
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;
/**
* Reads records that are delimited by a specifc begin/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) {
XmlRecordReader reader=null;
try {
reader = new XmlRecordReader((FileSplit) is, tac);
} catch (UnsupportedEncodingException ex) {
Logger.getLogger(XmlInputFormat.class.getName()).log(Level.SEVERE, null, ex);
} catch (IOException ex) {
Logger.getLogger(XmlInputFormat.class.getName()).log(Level.SEVERE, null, ex);
}
return reader;
}
public static class XmlRecordReader extends RecordReader<LongWritable,Text> {
private final byte[] startTag;
private final byte[] endTag;
private final long start;
private final long end;
private final FSDataInputStream fsin;
private final DataOutputBuffer buffer = new DataOutputBuffer();
private XmlRecordReader(FileSplit fileSplit, TaskAttemptContext tac) throws UnsupportedEncodingException, IOException {
System.out.println("Constructor=XMLRecordReader");
startTag = tac.getConfiguration().get(START_TAG_KEY).getBytes("utf-8");
endTag = tac.getConfiguration().get(END_TAG_KEY).getBytes("utf-8");
System.out.println("StartTag="+startTag+"---"+tac.getConfiguration().get(START_TAG_KEY));
System.out.println("EndTag="+endTag+"----"+tac.getConfiguration().get(END_TAG_KEY));
start = fileSplit.getStart();
end = start + fileSplit.getLength();
Path file = fileSplit.getPath();
FileSystem fs = file.getFileSystem(tac.getConfiguration());
fsin = fs.open(fileSplit.getPath());
fsin.seek(start);
}
@Override
public void initialize(InputSplit is, TaskAttemptContext tac) throws IOException, InterruptedException {
//FileSplit fileSplit= (FileSplit) is;
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
if (fsin.getPos() < end) {
if (readUntilMatch(startTag, false)) {
try {
buffer.write(startTag);
if (readUntilMatch(endTag, true)) {
//System.out.println();
// key.set(fsin.getPos());
// value.set(buffer.getData(), 0, buffer.getLength());
return true;
}
} finally {
buffer.reset();
}
}
}
return false;
}
@Override
public LongWritable getCurrentKey() throws IOException, InterruptedException {
return new LongWritable();
}
@Override
public Text getCurrentValue() throws IOException, InterruptedException {
byte[]val = buffer.getData();
System.out.println(val);
return new Text(val);
}
@Override
public float getProgress() throws IOException, InterruptedException {
return (fsin.getPos() - start) / (float) (end - start);
}
@Override
public void close() throws IOException {
fsin.close();
}
private boolean readUntilMatch(byte[] match, boolean withinBlock) throws IOException {
int i = 0;
while (true) {
int b = fsin.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 && fsin.getPos() >= end) return false;
}
}
}
}
