我会去做一个预处理工作，用xml标记异常。下一步你可以使用 XMLInputformat 处理文件(这只是解决方案的开始，根据您的反馈，我们可能会让事情变得更具体）
这个链接提供了一个编写您自己的xmlinputformat的教程，您可以自定义它来查找“异常”特征。本教程的要点是这样一句话：
如果记录跨越inputsplit边界，那么记录读取器将处理这个问题，这样我们就不必担心这个问题。
我会复制粘贴网站的信息，因为它可能会在未来离线，这可能会让人非常沮丧，在未来审查这一点：
输入格式：

package org.undercloud.mapreduce.example3;

import java.io.IOException;

import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;

public class XmlInputFormat extends FileInputFormat {

 public RecordReader getRecordReader(InputSplit input, JobConf job, Reporter reporter)
 throws IOException {

reporter.setStatus(input.toString());
 return new XmlRecordReader(job, (FileSplit)input);
 }

记录读取器：注意：读取拆分结束后的逻辑是 readUntilMatch 函数，如果存在打开的标记，则读取超过拆分结尾的内容。我想这才是你真正想要的！

package org.undercloud.mapreduce.example3;

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.*;
import org.apache.hadoop.mapred.*;

public class XmlRecordReader implements RecordReader {

    private String startTagS = "";
    private String endTagS = "";
    private byte[] startTag;
    private byte[] endTag;
    private long start;
    private long end;
    private FSDataInputStream fsin;
    private DataOutputBuffer buffer = new DataOutputBuffer();
    private LineRecordReader lineReader;
    private LongWritable lineKey;
    private Text lineValue;

    public XmlRecordReader(JobConf job, FileSplit split) throws IOException {
      lineReader = new LineRecordReader(job, split);
      lineKey = lineReader.createKey();
      lineValue = lineReader.createValue();
      startTag = startTagS.getBytes();
      endTag = endTagS.getBytes();

      // Open the file and seek to the start of the split
      start = split.getStart();
      end = start + split.getLength();
      Path file = split.getPath();
      FileSystem fs = file.getFileSystem(job);
      fsin = fs.open(split.getPath());
      fsin.seek(start);
   }

    public boolean next(Text key, XmlContent value) throws IOException {
    // Get the next line
        if (fsin.getPos() < end) { 
            if (readUntilMatch(startTag, false)) { 
                try { 
                    buffer.write(startTag); 
                    if (readUntilMatch(endTag, true)) { 
                        key.set(Long.toString(fsin.getPos())); 
                        value.bufferData = buffer.getData(); 
                        value.offsetData = 0; 
                        value.lenghtData = buffer.getLength(); 
                        return true; 
                    } 
                } 
                finally { 
                    buffer.reset(); 
                } 
            } 
        } 
        return false; 
    } 

    private boolean readUntilMatch(byte[] match, boolean withinBlock) throws IOException { 
        int i = 0; 
        while (true) { 
            int b = fsin.read(); // End of file -> T
            if (b == -1) return false;
            // F-> Save to buffer:
            if (withinBlock) buffer.write(b);
            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;
        }
    }

  public Text createKey() {
     return new Text("");
  }

  public XmlContent createValue() {
    return new XmlContent();
  }

  public long getPos() throws IOException {
    return lineReader.getPos();
  }

  public void close() throws IOException {
    lineReader.close();
  }

  public float getProgress() throws IOException {
    return lineReader.getProgress();
  }
}

最后是可写的：

package org.undercloud.mapreduce.example3;

import java.io.*;

import org.apache.hadoop.io.*;

public class XmlContent implements Writable{

    public byte[] bufferData;
    public int offsetData;
    public int lenghtData;

      public XmlContent(byte[] bufferData, int offsetData, int lenghtData) {
          this.bufferData = bufferData;
          this.offsetData = offsetData;
          this.lenghtData = lenghtData;
          }

      public XmlContent(){
          this(null,0,0);
      }

      public void write(DataOutput out) throws IOException {
          out.write(bufferData);
          out.writeInt(offsetData);
          out.writeInt(lenghtData);
      }

      public void readFields(DataInput in) throws IOException {
          in.readFully(bufferData);
          offsetData = in.readInt();
          lenghtData = in.readInt();
          }

      public String toString() {
            return Integer.toString(offsetData) + ", "
                + Integer.toString(lenghtData) +", "
                + bufferData.toString();
          }   

}

这看起来是一个非常有用的教程，解决了跨多个拆分的记录问题。让我知道如果你能适应你的问题这个例子。

类textinputformat和nlineinputformat可能会有所帮助。textinputformat将按行分割文件，因此如果异常以换行符结束（并且其中不包含任何内容），则应该可以这样做。如果异常包含固定行数，则nlineinputformat类应该是您想要的，因为您可以设置要采用的行数。
不幸的是，如果异常中可以包含数量可变的换行符，这将不起作用。
在这种情况下，我建议寻找mahout的xmlinputformat。它跨越了分裂的界限，所以对大多数东西都有效。只需运行一个预处理器，将异常放入 <exception></exception> 标记，并将其指定为开始/结束标记。
示例预处理器，使用regex识别异常

String input; //code this to the input string
String regex; //make this equal to the exception regex
BufferedWriter bw; //make this go to file where output will be stored

String toProcess = input;
boolean continueLoop = true;

while(continueLoop){
    Pattern p = Pattern.compile(regex);
    Matcher m = p.matcher(toProcess);
    if(m.find()){
        bw.write("<exception>"+toProcess.substring(m.start(),m.end())+"</exception>");
        toProcess = toProcess.substring(m.end());
    }else{
        continueLoop = false;
    }

}

谢谢你的解决方案。我认为这对我很有用
特别注意上面的评论
“如果记录跨越inputsplit边界，则记录读取器将处理此问题，因此我们不必担心此问题。”
然后我研究了linerecordreader如何读取数据表单的源代码。然后我发现实际上linerecordreader已经有了一些逻辑来读取inputsplit边界上的记录，因为由于块的大小限制，分割底部的行记录总是被分割成两个不同的分割。因此，我认为我需要做的是添加linerecordreader读取的数据大小。
现在我的解决方案是：重写linerecordreader中的方法“nextkeyvalue（）”。

public boolean nextKeyValue() throws IOException {
if (key == null) {
  key = new LongWritable();
}
key.set(pos);
if (value == null) {
  value = new Text();
}
int newSize = 0;
while (pos < end) {
  newSize = in.readLine(value, maxLineLength,
                        Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
                                 maxLineLength));

将行“while（pos<end）”更改为“while（pos<end+{param}）”
{param}表示readrecorder跨越分割边界读取的冗余数据的大小。