我正在尝试编写一个自定义读取器,用于读取具有定义字段数的记录(位于两行中)。
例如
1,2,3,4("," can be there or not)
,5,6,7,8
我的要求是读取记录并将其作为单个记录推入mapper,如 {1,2,3,4,5,6,7,8}
. 请提供一些信息。
更新:
public boolean nextKeyValue() throws IOException, InterruptedException {
if(key == null) {
key = new LongWritable();
}
//Current offset is the key
key.set(pos);
if(value == null) {
value = new Text();
}
int newSize = 0;
int numFields = 0;
Text temp = new Text();
boolean firstRead = true;
while(numFields < reqFields) {
while(pos < end) {
//Read up to the '\n' character and store it in 'temp'
newSize = in.readLine( temp,
maxLineLength,
Math.max((int) Math.min(Integer.MAX_VALUE, end - pos),
maxLineLength));
//If 0 bytes were read, then we are at the end of the split
if(newSize == 0) {
break;
}
//Otherwise update 'pos' with the number of bytes read
pos += newSize;
//If the line is not too long, check number of fields
if(newSize < maxLineLength) {
break;
}
//Line too long, try again
LOG.info("Skipped line of size " + newSize + " at pos " +
(pos - newSize));
}
//Exit, since we're at the end of split
if(newSize == 0) {
break;
}
else {
String record = temp.toString();
StringTokenizer fields = new StringTokenizer(record,"|");
numFields += fields.countTokens();
//Reset 'value' if this is the first append
if(firstRead) {
value = new Text();
firstRead = false;
}
if(numFields != reqFields) {
value.append(temp.getBytes(), 0, temp.getLength());
}
else {
value.append(temp.getBytes(), 0, temp.getLength());
}
}
}
if(newSize == 0) {
key = null;
value = null;
return false;
}
else {
return true;
}
}
}
这是我正在尝试的nextkeyvalue方法。但是Map器仍然没有得到正确的值。reqfields是4。
2条答案
按热度按时间fafcakar1#
看看textinputformat是如何实现的。看看它的超类,还有fileinputformat。必须将textinputformat子类化为fileinputformat,并实现自己的记录处理。
在实现任何类型的文件输入格式时需要注意的是:
framework将分割文件,并给出要读取的文件片段的起始偏移量和字节长度。它很可能会将文件直接拆分到某个记录上。这就是为什么如果拆分中没有完全包含该记录,则读取器必须在拆分开始时跳过该记录的字节;如果拆分中没有完全包含该记录,则读取器必须读入拆分的最后一个字节以读取整个最后一条记录。
例如,textinoutformat将\n字符视为记录分隔符,因此当它获得拆分时,它将跳过字节直到第一个\n字符,并在拆分结束后读取到第\n个字符。
至于代码示例:
你需要问自己以下问题:假设你打开文件,找到一个随机的位置,开始向前阅读。如何检测记录的开始?我在您的代码中看不到任何与此相关的内容,没有它,您就无法编写好的输入格式,因为您不知道记录边界是什么。
现在仍然可以通过使issplittable(jobcontext,path)方法返回false,使输入格式从端到端读取整个文件。这使得文件完全由单个map任务读取,从而降低了并行性。
您的内部while循环似乎有问题,因为它正在检查过长的行并跳过它们。假设您的记录是使用多行写入的,那么在读取时可能会合并一条记录的一部分和另一条记录的另一部分。
6qftjkof2#
字符串必须使用stringtokenizer进行标记化,而不是拆分。代码已更新为新的实现。