StackOverflow2013

Note that there are some explanatory texts on larger screens.

plurals

POError on Amazon aws EMR while running Mapreduce code
text
Body
copied!<p>I am writing a map reduce code on my local cluster to Parse XML file. I am using jdom parser for that. It is working fine on my local cluster.</p> <p>But when I am trying to create a job flow on EMR it is giving me following error :</p> <p>" Error: Found class org.apache.hadoop.mapreduce.TaskAttemptContext, but interface was expected "</p> <p>I am using Hadoop 0.20.2 on my local machine. Please let me know if anyone knows what is the issue.</p> <p>Here is the code for the record reader:</p> <pre><code>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; /** * 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) { 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 fsin; 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()); fsin = fs.open(fileSplit.getPath()); fsin.seek(start); } @Override public boolean nextKeyValue() throws IOException, InterruptedException { if (fsin.getPos() < end) { if (readUntilMatch(startTag, false)) { try { buffer.write(startTag); if (readUntilMatch(endTag, true)) { value.set(buffer.getData(), 0, buffer.getLength()); key.set(fsin.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 (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; } } } } </code></pre>

Querying!

Guidance

An individual column

Larger individual text columns get their own page to allow for proper reading.

SQuiL has stopped working due to an internal error.

If you are curious you may find further information in the browser console, which is accessible through the devtools (F12).

Reload