小编给大家分享一下如何实现RecordReader按行读取，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！
成都创新互联公司是一家专业提供铁锋企业网站建设,专注与成都做网站、成都网站建设、成都外贸网站建设、H5建站、小程序制作等业务。10年已为铁锋众多企业、政府机构等服务。创新互联专业网站设计公司优惠进行中。
public class CustomLineRecordReader
extends RecordReader {
private long start;
private long pos;
private long end;
private LineReader in;
private int maxLineLength;
private LongWritable key = new LongWritable();
private Text value = new Text();
private static final Log LOG = LogFactory.getLog(
CustomLineRecordReader.class);
/**
* From Design Pattern, O'Reilly...
* This method takes as arguments the map task’s assigned InputSplit and
* TaskAttemptContext, and prepares the record reader. For file-based input
* formats, this is a good place to seek to the byte position in the file to
* begin reading.
*/
@Override
public void initialize(
InputSplit genericSplit,
TaskAttemptContext context)
throws IOException {
// This InputSplit is a FileInputSplit
FileSplit split = (FileSplit) genericSplit;
// Retrieve configuration, and Max allowed
// bytes for a single record
Configuration job = context.getConfiguration();
this.maxLineLength = job.getInt(
"mapred.linerecordreader.maxlength",
Integer.MAX_VALUE);
// Split "S" is responsible for all records
// starting from "start" and "end" positions
start = split.getStart();
end = start + split.getLength();
// Retrieve file containing Split "S"
final Path file = split.getPath();
FileSystem fs = file.getFileSystem(job);
FSDataInputStream fileIn = fs.open(split.getPath());
// If Split "S" starts at byte 0, first line will be processed
// If Split "S" does not start at byte 0, first line has been already
// processed by "S-1" and therefore needs to be silently ignored
boolean skipFirstLine = false;
if (start != 0) {
skipFirstLine = true;
// Set the file pointer at "start - 1" position.
// This is to make sure we won't miss any line
// It could happen if "start" is located on a EOL
--start;
fileIn.seek(start);
}
in = new LineReader(fileIn, job);
// If first line needs to be skipped, read first line
// and stores its content to a dummy Text
if (skipFirstLine) {
Text dummy = new Text();
// Reset "start" to "start + line offset"
start += in.readLine(dummy, 0,
(int) Math.min(
(long) Integer.MAX_VALUE,
end - start));
}
// Position is the actual start
this.pos = start;
}
/**
* From Design Pattern, O'Reilly...
* Like the corresponding method of the InputFormat class, this reads a
* single key/ value pair and returns true until the data is consumed.
*/
@Override
public boolean nextKeyValue() throws IOException {
// Current offset is the key
key.set(pos);
int newSize = 0;
// Make sure we get at least one record that starts in this Split
while (pos < end) {
// Read first line and store its content to "value"
newSize = in.readLine(value, maxLineLength,
Math.max((int) Math.min(
Integer.MAX_VALUE, end - pos),
maxLineLength));
// No byte read, seems that we reached end of Split
// Break and return false (no key / value)
if (newSize == 0) {
break;
}
// Line is read, new position is set
pos += newSize;
// Line is lower than Maximum record line size
// break and return true (found key / value)
if (newSize < maxLineLength) {
break;
}
// Line is too long
// Try again with position = position + line offset,
// i.e. ignore line and go to next one
// TODO: Shouldn't it be LOG.error instead ??
LOG.info("Skipped line of size " +
newSize + " at pos "
+ (pos - newSize));
}
if (newSize == 0) {
// We've reached end of Split
key = null;
value = null;
return false;
} else {
// Tell Hadoop a new line has been found
// key / value will be retrieved by
// getCurrentKey getCurrentValue methods
return true;
}
}
/**
* From Design Pattern, O'Reilly...
* This methods are used by the framework to give generated key/value pairs
* to an implementation of Mapper. Be sure to reuse the objects returned by
* these methods if at all possible!
*/
@Override
public LongWritable getCurrentKey() throws IOException,
InterruptedException {
return key;
}
/**
* From Design Pattern, O'Reilly...
* This methods are used by the framework to give generated key/value pairs
* to an implementation of Mapper. Be sure to reuse the objects returned by
* these methods if at all possible!
*/
@Override
public Text getCurrentValue() throws IOException, InterruptedException {
return value;
}
/**
* From Design Pattern, O'Reilly...
* Like the corresponding method of the InputFormat class, this is an
* optional method used by the framework for metrics gathering.
*/
@Override
public float getProgress() throws IOException, InterruptedException {
if (start == end) {
return 0.0f;
} else {
return Math.min(1.0f, (pos - start) / (float) (end - start));
}
}
/**
* From Design Pattern, O'Reilly...
* This method is used by the framework for cleanup after there are no more
* key/value pairs to process.
*/
@Override
public void close() throws IOException {
if (in != null) {
in.close();
}
}
}
以上是“如何实现RecordReader按行读取”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注创新互联行业资讯频道！
当前文章：如何实现RecordReader按行读取
分享URL：
http://cdxtjz.cn/article/ipsgjh.html
