hadoop-hdfs-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Mohammad Tariq <donta...@gmail.com>
Subject Re: Processing xml documents using StreamXmlRecordReader
Date Thu, 21 Jun 2012 07:12:42 GMT
Hello Madhu,

           I really appreciate your efforts. I am sorry I did not
respond back.Actually I was struggling with it, so had anything to let
you know.Many thanks.

Regards,
    Mohammad Tariq


On Thu, Jun 21, 2012 at 12:37 PM, madhu phatak <phatak.dev@gmail.com> wrote:
> Hi,
>  Jira for the new API code
> https://issues.apache.org/jira/browse/HADOOP-8521
>
>
> On Tue, Jun 19, 2012 at 6:11 PM, madhu phatak <phatak.dev@gmail.com> wrote:
>>
>> Hi,
>>  Yes you have the class, but it's for old API.
>>
>>  Please find the code below  for ported classes for new API. I have not
>> tested the code,try to use these classes and let me know if its working for
>> you.
>>
>>
>> StreamInputFormat (new API)
>>
>> package org.apache.hadoop.streaming;
>> import java.io.IOException;
>> import java.lang.reflect.Constructor;
>>
>> import org.apache.hadoop.conf.Configuration;
>> import org.apache.hadoop.fs.FSDataInputStream;
>> import org.apache.hadoop.fs.FileSystem;
>> import org.apache.hadoop.io.Text;
>> import org.apache.hadoop.mapred.JobConf;
>> import org.apache.hadoop.mapred.Reporter;
>> 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.KeyValueTextInputFormat;
>> import org.apache.hadoop.streaming.StreamUtil;
>>
>> /**
>>  * 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.
>>  */
>>
>>
>>
>>
>> /** An input format that selects a RecordReader based on a JobConf
>> property.
>>  *  This should be used only for non-standard record reader such as
>>  *  StreamXmlRecordReader. For all other standard
>>  *  record readers, the appropriate input format classes should be used.
>>  */
>> public class StreamInputFormat extends KeyValueTextInputFormat {
>>
>> @Override
>> public RecordReader<Text, Text> createRecordReader(InputSplit
>> genericSplit,
>> TaskAttemptContext context) throws IOException {
>>
>> Configuration conf = context.getConfiguration();
>> String c = conf.get("stream.recordreader.class");
>> if (c == null || c.indexOf("LineRecordReader") >= 0) {
>> return super.createRecordReader(genericSplit, context);
>> }
>>
>> // handling non-standard record reader (likely StreamXmlRecordReader)
>> FileSplit split = (FileSplit) genericSplit;
>> //LOG.info("getRecordReader start.....split=" + split);
>> context.setStatus(split.toString());
>> context.progress();
>>
>> // Open the file and seek to the start of the split
>> FileSystem fs = split.getPath().getFileSystem(conf);
>> FSDataInputStream in = fs.open(split.getPath());
>>
>> // Factory dispatch based on available params..
>> Class readerClass;
>>
>> {
>> readerClass = StreamUtil.goodClassOrNull(conf, c, null);
>> if (readerClass == null) {
>> throw new RuntimeException("Class not found: " + c);
>> }
>> }
>> Constructor ctor;
>>    try {
>>      ctor = readerClass.getConstructor(new Class[] {
>> FSDataInputStream.class,
>>                                                      FileSplit.class,
>> TaskAttemptContext.class, Configuration.class, FileSystem.class });
>>    } catch (NoSuchMethodException nsm) {
>>      throw new RuntimeException(nsm);
>>    }
>>
>>    RecordReader<Text, Text> reader;
>>    try {
>>      reader = (RecordReader<Text, Text>) ctor.newInstance(new Object[] {
>> in, split,
>>                                                      
       context,
>> conf, fs });
>>    } catch (Exception nsm) {
>>      throw new RuntimeException(nsm);
>>    }
>>    return reader;
>>
>>
>> }
>>
>> }
>>
>> StreamXmlRecordReader
>>
>> /**
>>  * 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.
>>  */
>>
>> package org.apache.hadoop.streaming;
>>
>> import java.io.*;
>> import java.util.regex.*;
>>
>> import org.apache.hadoop.io.DataOutputBuffer;
>> import org.apache.hadoop.io.Writable;
>> import org.apache.hadoop.io.Text;
>> import org.apache.hadoop.io.WritableComparable;
>> import org.apache.hadoop.conf.Configuration;
>> import org.apache.hadoop.fs.FileSystem;
>> import org.apache.hadoop.fs.FSDataInputStream;
>> import org.apache.hadoop.mapred.Reporter;
>> import org.apache.hadoop.mapred.FileSplit;
>> import org.apache.hadoop.mapred.JobConf;
>> import org.apache.hadoop.mapreduce.TaskAttemptContext;
>>
>> /** A way to interpret XML fragments as Mapper input records.
>>  *  Values are XML subtrees delimited by configurable tags.
>>  *  Keys could be the value of a certain attribute in the XML subtree,
>>  *  but this is left to the stream processor application.
>>  *
>>  *  The name-value properties that StreamXmlRecordReader understands are:
>>  *    String begin (chars marking beginning of record)
>>  *    String end   (chars marking end of record)
>>  *    int maxrec   (maximum record size)
>>  *    int lookahead(maximum lookahead to sync CDATA)
>>  *    boolean slowmatch
>>  */
>> public class StreamXmlRecordReader extends StreamBaseRecordReader {
>>
>>   public StreamXmlRecordReader(FSDataInputStream in, FileSplit split,
>> TaskAttemptContext context,
>>                                Configuration conf, FileSystem fs)
throws
>> IOException {
>>     super(in, split, context, conf, fs);
>>
>>     beginMark_ = checkJobGet(CONF_NS + "begin");
>>     endMark_ = checkJobGet(CONF_NS + "end");
>>
>>     maxRecSize_ = conf_.getInt(CONF_NS + "maxrec", 50 * 1000);
>>     lookAhead_ = conf_.getInt(CONF_NS + "lookahead", 2 * maxRecSize_);
>>     synched_ = false;
>>
>>     slowMatch_ = conf_.getBoolean(CONF_NS + "slowmatch", false);
>>     if (slowMatch_) {
>>       beginPat_ = makePatternCDataOrMark(beginMark_);
>>       endPat_ = makePatternCDataOrMark(endMark_);
>>     }
>>     init();
>>   }
>>
>>   public void init() throws IOException {
>>     LOG.info("StreamBaseRecordReader.init: " + " start_=" + start_ + "
>> end_=" + end_ + " length_="
>>              + length_ + " start_ > in_.getPos() =" + (start_ >
>> in_.getPos()) + " " + start_ + " > "
>>              + in_.getPos());
>>     if (start_ > in_.getPos()) {
>>       in_.seek(start_);
>>     }
>>     pos_ = start_;
>>     bin_ = new BufferedInputStream(in_);
>>     seekNextRecordBoundary();
>>   }
>>
>>   int numNext = 0;
>>
>>   public synchronized boolean next(Text key, Text value) throws
>> IOException {
>>     numNext++;
>>     if (pos_ >= end_) {
>>       return false;
>>     }
>>
>>     DataOutputBuffer buf = new DataOutputBuffer();
>>     if (!readUntilMatchBegin()) {
>>       return false;
>>     }
>>     if (pos_ >= end_ || !readUntilMatchEnd(buf)) {
>>       return false;
>>     }
>>
>>     // There is only one elem..key/value splitting is not done here.
>>     byte[] record = new byte[buf.getLength()];
>>     System.arraycopy(buf.getData(), 0, record, 0, record.length);
>>
>>     numRecStats(record, 0, record.length);
>>
>>     key.set(record);
>>     value.set("");
>>
>>     return true;
>>   }
>>
>>   public void seekNextRecordBoundary() throws IOException {
>>     readUntilMatchBegin();
>>   }
>>
>>   boolean readUntilMatchBegin() throws IOException {
>>     if (slowMatch_) {
>>       return slowReadUntilMatch(beginPat_, false, null);
>>     } else {
>>       return fastReadUntilMatch(beginMark_, false, null);
>>     }
>>   }
>>
>>   private boolean readUntilMatchEnd(DataOutputBuffer buf) throws
>> IOException {
>>     if (slowMatch_) {
>>       return slowReadUntilMatch(endPat_, true, buf);
>>     } else {
>>       return fastReadUntilMatch(endMark_, true, buf);
>>     }
>>   }
>>
>>   private boolean slowReadUntilMatch(Pattern markPattern, boolean
>> includePat,
>>                                      DataOutputBuffer outBufOrNull)
throws
>> IOException {
>>     byte[] buf = new byte[Math.max(lookAhead_, maxRecSize_)];
>>     int read = 0;
>>     bin_.mark(Math.max(lookAhead_, maxRecSize_) + 2); //mark to invalidate
>> if we read more
>>     read = bin_.read(buf);
>>     if (read == -1) return false;
>>
>>     String sbuf = new String(buf, 0, read, "UTF-8");
>>     Matcher match = markPattern.matcher(sbuf);
>>
>>     firstMatchStart_ = NA;
>>     firstMatchEnd_ = NA;
>>     int bufPos = 0;
>>     int state = synched_ ? CDATA_OUT : CDATA_UNK;
>>     int s = 0;
>>
>>     while (match.find(bufPos)) {
>>       int input;
>>       if (match.group(1) != null) {
>>         input = CDATA_BEGIN;
>>       } else if (match.group(2) != null) {
>>         input = CDATA_END;
>>         firstMatchStart_ = NA; // |<DOC CDATA[ </DOC> ]]> should
keep it
>>       } else {
>>         input = RECORD_MAYBE;
>>       }
>>       if (input == RECORD_MAYBE) {
>>         if (firstMatchStart_ == NA) {
>>           firstMatchStart_ = match.start();
>>           firstMatchEnd_ = match.end();
>>         }
>>       }
>>       state = nextState(state, input, match.start());
>>       if (state == RECORD_ACCEPT) {
>>         break;
>>       }
>>       bufPos = match.end();
>>       s++;
>>     }
>>     if (state != CDATA_UNK) {
>>       synched_ = true;
>>     }
>>     boolean matched = (firstMatchStart_ != NA) && (state == RECORD_ACCEPT
>> || state == CDATA_UNK);
>>     if (matched) {
>>       int endPos = includePat ? firstMatchEnd_ : firstMatchStart_;
>>       bin_.reset();
>>
>>       for (long skiplen = endPos; skiplen > 0; ) {
>>         skiplen -= bin_.skip(skiplen); // Skip succeeds as we have read
>> this buffer
>>       }
>>
>>       pos_ += endPos;
>>       if (outBufOrNull != null) {
>>         outBufOrNull.writeBytes(sbuf.substring(0,endPos));
>>       }
>>     }
>>     return matched;
>>   }
>>
>>   // states
>>   final static int CDATA_IN = 10;
>>   final static int CDATA_OUT = 11;
>>   final static int CDATA_UNK = 12;
>>   final static int RECORD_ACCEPT = 13;
>>   // inputs
>>   final static int CDATA_BEGIN = 20;
>>   final static int CDATA_END = 21;
>>   final static int RECORD_MAYBE = 22;
>>
>>   /* also updates firstMatchStart_;*/
>>   int nextState(int state, int input, int bufPos) {
>>     switch (state) {
>>     case CDATA_UNK:
>>     case CDATA_OUT:
>>       switch (input) {
>>       case CDATA_BEGIN:
>>         return CDATA_IN;
>>       case CDATA_END:
>>         if (state == CDATA_OUT) {
>>           //System.out.println("buggy XML " + bufPos);
>>         }
>>         return CDATA_OUT;
>>       case RECORD_MAYBE:
>>         return (state == CDATA_UNK) ? CDATA_UNK : RECORD_ACCEPT;
>>       }
>>       break;
>>     case CDATA_IN:
>>       return (input == CDATA_END) ? CDATA_OUT : CDATA_IN;
>>     }
>>     throw new IllegalStateException(state + " " + input + " " + bufPos + "
>> " + splitName_);
>>   }
>>
>>   Pattern makePatternCDataOrMark(String escapedMark) {
>>     StringBuffer pat = new StringBuffer();
>>     addGroup(pat, StreamUtil.regexpEscape("CDATA[")); // CDATA_BEGIN
>>     addGroup(pat, StreamUtil.regexpEscape("]]>")); // CDATA_END
>>     addGroup(pat, escapedMark); // RECORD_MAYBE
>>     return Pattern.compile(pat.toString());
>>   }
>>
>>   void addGroup(StringBuffer pat, String escapedGroup) {
>>     if (pat.length() > 0) {
>>       pat.append("|");
>>     }
>>     pat.append("(");
>>     pat.append(escapedGroup);
>>     pat.append(")");
>>   }
>>
>>   boolean fastReadUntilMatch(String textPat, boolean includePat,
>> DataOutputBuffer outBufOrNull) throws IOException {
>>     byte[] cpat = textPat.getBytes("UTF-8");
>>     int m = 0;
>>     boolean match = false;
>>     int msup = cpat.length;
>>     int LL = 120000 * 10;
>>
>>     bin_.mark(LL); // large number to invalidate mark
>>     while (true) {
>>       int b = bin_.read();
>>       if (b == -1) break;
>>
>>       byte c = (byte) b; // this assumes eight-bit matching. OK with UTF-8
>>       if (c == cpat[m]) {
>>         m++;
>>         if (m == msup) {
>>           match = true;
>>           break;
>>         }
>>       } else {
>>         bin_.mark(LL); // rest mark so we could jump back if we found a
>> match
>>         if (outBufOrNull != null) {
>>           outBufOrNull.write(cpat, 0, m);
>>           outBufOrNull.write(c);
>>         }
>>         pos_ += m + 1; // skip m chars, +1 for 'c'
>>         m = 0;
>>       }
>>     }
>>     if (!includePat && match) {
>>       bin_.reset();
>>     } else if (outBufOrNull != null) {
>>       outBufOrNull.write(cpat);
>>       pos_ += msup;
>>     }
>>     return match;
>>   }
>>
>>   String checkJobGet(String prop) throws IOException {
>>     String val = conf_.get(prop);
>>     if (val == null) {
>>       throw new IOException("JobConf: missing required property: " +
>> prop);
>>     }
>>     return val;
>>   }
>>
>>   String beginMark_;
>>   String endMark_;
>>
>>   Pattern beginPat_;
>>   Pattern endPat_;
>>
>>   boolean slowMatch_;
>>   int lookAhead_; // bytes to read to try to synch CDATA/non-CDATA. Should
>> be more than max record size
>>   int maxRecSize_;
>>
>>   BufferedInputStream bin_; // Wrap FSDataInputStream for efficient
>> backward seeks
>>   long pos_; // Keep track on position with respect encapsulated
>> FSDataInputStream
>>
>>   final static int NA = -1;
>>   int firstMatchStart_ = 0; // candidate record boundary. Might just be
>> CDATA.
>>   int firstMatchEnd_ = 0;
>>
>>   boolean synched_;
>> }
>>
>> StreamBaseRecordReader
>>
>> /**
>>  * 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.
>>  */
>>
>> package org.apache.hadoop.streaming;
>>
>> import java.io.*;
>>
>> import org.apache.hadoop.io.Text;
>> import org.apache.hadoop.io.Writable;
>> import org.apache.hadoop.io.WritableComparable;
>> import org.apache.hadoop.conf.Configuration;
>> import org.apache.hadoop.fs.Path;
>> import org.apache.hadoop.fs.FileSystem;
>> import org.apache.hadoop.fs.FSDataInputStream;
>> import org.apache.hadoop.mapred.Reporter;
>> import org.apache.hadoop.mapred.RecordReader;
>> import org.apache.hadoop.mapred.FileSplit;
>> import org.apache.hadoop.mapred.JobConf;
>> import org.apache.hadoop.mapreduce.TaskAttemptContext;
>> import org.apache.commons.logging.*;
>>
>> /**
>>  * Shared functionality for hadoopStreaming formats.
>>  * A custom reader can be defined to be a RecordReader with the
>> constructor below
>>  * and is selected with the option bin/hadoopStreaming -inputreader ...
>>  * @see StreamXmlRecordReader
>>  */
>> public abstract class StreamBaseRecordReader implements RecordReader<Text,
>> Text> {
>>
>>   protected static final Log LOG =
>> LogFactory.getLog(StreamBaseRecordReader.class.getName());
>>
>>   // custom JobConf properties for this class are prefixed with this
>> namespace
>>   final static String CONF_NS = "stream.recordreader.";
>>
>>   public StreamBaseRecordReader(FSDataInputStream in, FileSplit split,
>> TaskAttemptContext context,
>>                                 Configuration conf, FileSystem fs)
throws
>> IOException {
>>     in_ = in;
>>     split_ = split;
>>     start_ = split_.getStart();
>>     length_ = split_.getLength();
>>     end_ = start_ + length_;
>>     splitName_ = split_.getPath().getName();
>>     this.context_ = context;
>>     conf_ = conf;
>>     fs_ = fs;
>>
>>     statusMaxRecordChars_ = conf.getInt(CONF_NS + "statuschars", 200);
>>   }
>>
>>   /// RecordReader API
>>
>>   /** Read a record. Implementation should call numRecStats at the end
>>    */
>>   public abstract boolean next(Text key, Text value) throws IOException;
>>
>>   /** Returns the current position in the input. */
>>   public synchronized long getPos() throws IOException {
>>     return in_.getPos();
>>   }
>>
>>   /** Close this to future operations.*/
>>   public synchronized void close() throws IOException {
>>     in_.close();
>>   }
>>
>>   public float getProgress() throws IOException {
>>     if (end_ == start_) {
>>       return 1.0f;
>>     } else {
>>       return ((float)(in_.getPos() - start_)) / ((float)(end_ - start_));
>>     }
>>   }
>>
>>   public Text createKey() {
>>     return new Text();
>>   }
>>
>>   public Text createValue() {
>>     return new Text();
>>   }
>>
>>   /// StreamBaseRecordReader API
>>
>>   /** Implementation should seek forward in_ to the first byte of the next
>> record.
>>    *  The initial byte offset in the stream is arbitrary.
>>    */
>>   public abstract void seekNextRecordBoundary() throws IOException;
>>
>>   void numRecStats(byte[] record, int start, int len) throws IOException {
>>     numRec_++;
>>     if (numRec_ == nextStatusRec_) {
>>       String recordStr = new String(record, start, Math.min(len,
>> statusMaxRecordChars_), "UTF-8");
>>       nextStatusRec_ += 100;//*= 10;
>>       String status = getStatus(recordStr);
>>       LOG.info(status);
>>       context_.setStatus(status);
>>     }
>>   }
>>
>>   long lastMem = 0;
>>
>>   String getStatus(CharSequence record) {
>>     long pos = -1;
>>     try {
>>       pos = getPos();
>>     } catch (IOException io) {
>>     }
>>     String recStr;
>>     if (record.length() > statusMaxRecordChars_) {
>>       recStr = record.subSequence(0, statusMaxRecordChars_) + "...";
>>     } else {
>>       recStr = record.toString();
>>     }
>>     String unqualSplit = split_.getPath().getName() + ":" +
>>                          split_.getStart() + "+" + split_.getLength();
>>     String status = "HSTR " + StreamUtil.HOST + " " + numRec_ + ". pos=" +
>> pos + " " + unqualSplit
>>       + " Processing record=" + recStr;
>>     status += " " + splitName_;
>>     return status;
>>   }
>>
>>   FSDataInputStream in_;
>>   FileSplit split_;
>>   long start_;
>>   long end_;
>>   long length_;
>>   String splitName_;
>>   TaskAttemptContext context_;
>>   Configuration conf_;
>>   FileSystem fs_;
>>   int numRec_ = 0;
>>   int nextStatusRec_ = 1;
>>   int statusMaxRecordChars_;
>>
>> }
>>
>>
>>
>> On Tue, Jun 19, 2012 at 5:58 PM, Mohammad Tariq <dontariq@gmail.com>
>> wrote:
>>>
>>> But I have downloaded "hadoop-streaming-0.20.205.0.jar" and it
>>> contains StreamXmlRecordReader.class file. This means it should
>>> support StreamInputFormat.
>>>
>>> Regards,
>>>     Mohammad Tariq
>>>
>>>
>>> On Tue, Jun 19, 2012 at 5:54 PM, Mohammad Tariq <dontariq@gmail.com>
>>> wrote:
>>> > Thanks Madhu. I'll do that.
>>> >
>>> > Regards,
>>> >     Mohammad Tariq
>>> >
>>> >
>>> > On Tue, Jun 19, 2012 at 5:43 PM, madhu phatak <phatak.dev@gmail.com>
>>> > wrote:
>>> >> Seems like StreamInputFormat not yet ported to new API.That's why you
>>> >> are
>>> >> not able to set as InputFormatClass. You can file a  jira for this
>>> >> issue.
>>> >>
>>> >>
>>> >> On Tue, Jun 19, 2012 at 4:49 PM, Mohammad Tariq <dontariq@gmail.com>
>>> >> wrote:
>>> >>>
>>> >>> My driver function looks like this -
>>> >>>
>>> >>> public static void main(String[] args) throws IOException,
>>> >>> InterruptedException, ClassNotFoundException {
>>> >>>                // TODO Auto-generated method stub
>>> >>>
>>> >>>                Configuration conf = new Configuration();
>>> >>>                Job job = new Job();
>>> >>>                conf.set("stream.recordreader.class",
>>> >>> "org.apache.hadoop.streaming.StreamXmlRecordReader");
>>> >>>                conf.set("stream.recordreader.begin", "<info>");
>>> >>>                conf.set("stream.recordreader.end", "</info>");
>>> >>>                job.setInputFormatClass(StreamInputFormat.class);
>>> >>>                job.setOutputKeyClass(Text.class);
>>> >>>                job.setOutputValueClass(IntWritable.class);
>>> >>>                FileInputFormat.addInputPath(job, new
>>> >>> Path("/mapin/demo.xml"));
>>> >>>                FileOutputFormat.setOutputPath(job, new
>>> >>> Path("/mapout/demo"));
>>> >>>                job.waitForCompletion(true);
>>> >>>        }
>>> >>>
>>> >>> Could you please out my mistake??
>>> >>>
>>> >>> Regards,
>>> >>>     Mohammad Tariq
>>> >>>
>>> >>>
>>> >>> On Tue, Jun 19, 2012 at 4:35 PM, Mohammad Tariq <dontariq@gmail.com>
>>> >>> wrote:
>>> >>> > Hello Madhu,
>>> >>> >
>>> >>> >             Thanks for the response. Actually I was trying
to use
>>> >>> > the
>>> >>> > new API (Job). Have you tried that. I was not able to set the
>>> >>> > InputFormat using the Job API.
>>> >>> >
>>> >>> > Regards,
>>> >>> >     Mohammad Tariq
>>> >>> >
>>> >>> >
>>> >>> > On Tue, Jun 19, 2012 at 4:28 PM, madhu phatak
>>> >>> > <phatak.dev@gmail.com>
>>> >>> > wrote:
>>> >>> >> Hi,
>>> >>> >>  Set the following properties in driver class
>>> >>> >>
>>> >>> >>   jobConf.set("stream.recordreader.class",
>>> >>> >> "org.apache.hadoop.streaming.StreamXmlRecordReader");
>>> >>> >> jobConf.set("stream.recordreader.begin",
>>> >>> >> "start-tag");
>>> >>> >> jobConf.set("stream.recordreader.end",
>>> >>> >> "end-tag");
>>> >>> >>
>>> >>> >> jobConf.setInputFormat(StreamInputFormat,class);
>>> >>> >>
>>> >>> >>  In Mapper, xml record will come as key of type Text,so
your
>>> >>> >> mapper
>>> >>> >> will
>>> >>> >> look like
>>> >>> >>
>>> >>> >>   public class MyMapper<K,V>  implements Mapper<Text,Text,K,V>
>>> >>> >>
>>> >>> >>
>>> >>> >> On Tue, Jun 19, 2012 at 2:49 AM, Mohammad Tariq
>>> >>> >> <dontariq@gmail.com>
>>> >>> >> wrote:
>>> >>> >>>
>>> >>> >>> Hello list,
>>> >>> >>>
>>> >>> >>>        Could anyone, who has written MapReduce
jobs to process
>>> >>> >>> xml
>>> >>> >>> documents stored in there cluster using "StreamXmlRecordReader"
>>> >>> >>> share
>>> >>> >>> his/her experience??...or if you can provide me some
pointers
>>> >>> >>> addressing that..Many thanks.
>>> >>> >>>
>>> >>> >>> Regards,
>>> >>> >>>     Mohammad Tariq
>>> >>> >>
>>> >>> >>
>>> >>> >>
>>> >>> >>
>>> >>> >> --
>>> >>> >> https://github.com/zinnia-phatak-dev/Nectar
>>> >>> >>
>>> >>
>>> >>
>>> >>
>>> >>
>>> >> --
>>> >> https://github.com/zinnia-phatak-dev/Nectar
>>> >>
>>
>>
>>
>>
>> --
>> https://github.com/zinnia-phatak-dev/Nectar
>>
>
>
>
> --
> https://github.com/zinnia-phatak-dev/Nectar
>

Mime
View raw message