我想使用 hadoop 从原始日志中获取和解析异常。我遇到一个问题，一些异常（跨越多行）将是 2 个不同拆分的一部分，因此是 2 个不同的映射器。
我有一个避免这个问题的想法。我可以重写该getSplits()方法以使每个拆分都有一些冗余数据。我认为这个解决方案对我来说成本太高了。
那么有人对这个问题有更好的解决方案吗？
问问题
270 次
3 回答
2
我会去做一个预处理工作，用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函数中，如果存在 open tag ，则读取拆分结束后。我想这真的是你要找的东西！
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();
}
}
这看起来像是一个非常有用的教程，解决了跨越多个拆分的记录问题。如果您能够根据您的问题调整此示例，请告诉我。
于 2013-11-04T08:03:52.060 回答
1
TextInputFormat和NLineInputFormat类可能会有所帮助。TextInputFormat 将逐行拆分文件，因此如果异常以换行符结尾（并且其中不包含任何内容），这应该可以工作。如果异常包含固定数量的行，则 NLineInputFormat 类应该是您想要的，因为您可以设置要采用的行数。
不幸的是，如果异常中可以包含可变数量的换行符，这将不起作用。
在这种情况下，我建议寻找 Mahout 的XmlInputFormat。它跨越了分裂的界限，因此适用于大多数东西。只需运行预处理器将异常放入<exception></exception>标签中，并将其指定为开始/结束标签。
预处理器示例，使用正则表达式识别异常
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;
}
}
于 2013-10-29T13:50:56.197 回答
0
感谢您的所有解决方案。我觉得对我有用
特别注意上面的评论
“如果记录跨越 InputSplit 边界，记录读取器将处理此问题，因此我们不必担心这一点。”
然后我查看了有关 LineRecordReader 如何读取数据表单拆分的源代码。然后我发现实际上 LineRecordReader 已经有一些逻辑来读取跨越 InputSplit 边界的记录，因为由于块的大小限制，拆分底部的行记录总是被拆分为 2 个不同的拆分。所以我认为我需要做的是添加 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})”</p>
{param} 表示 readRecorder 跨分割边界读取的冗余数据的大小。
于 2013-11-14T03:24:54.380 回答