Hadoop中map()方法的接口和源代码以及操作示例是什么?
● public void map(Text relPath,CopyListingFileStatus sourceFileStatus, Context context) throws IOException,InterruptedException
从输入参数可以看出来,这其实就是对上面分析过的UniformSizeInputFormat类里分片后的数据里的每一行进行处理,每行里存放的就是“K-V”对,也就是fileList.seq文件每行的内容。Map方法体前半部分有一大堆代码内容,其实都是一些准备和判断工作,为后面的拷贝服务,最终的拷贝动作发生在copyFileWithRetry(description,sourceCurrStatus, target, context, action, fileAttributes)这个函数中,进入这个函数一直往里面读,就能看到数据最终通过常用的Java输入输出流的方式完成点对点拷贝,最后拷贝方法源码如下:
long copyBytes(FileStatus sourceFileStatus, long sourceOffset,
OutputStream outStream, int bufferSize, Mapper.Context context)
throws IOException {
Path source = sourceFileStatus.getPath();
byte buf[] = new byte[bufferSize];
ThrottledInputStream inStream = null;
long totalBytesRead = 0;
try {
inStream = getInputStream(source, context.getConfiguration());
int bytesRead = readBytes(inStream, buf, sourceOffset);
while (bytesRead >= 0) {
totalBytesRead += bytesRead;
if (action == FileAction.APPEND) {
sourceOffset += bytesRead;
}
outStream.write(buf, 0, bytesRead);
updateContextStatus(totalBytesRead, context, sourceFileStatus);
bytesRead = readBytes(inStream, buf, sourceOffset);
}
outStream.close();
outStream = null;
} finally {
IOUtils.cleanup(LOG, outStream, inStream);
}
return totalBytesRead;
}
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。