Hadoop MapReduce中如何处理跨行Block和inputSplit

2024-08-25 06:48

本文主要是介绍Hadoop MapReduce中如何处理跨行Block和inputSplit,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

Hadoop MapReduce中如何处理跨行Block和inputSplit
http://www.aboutyun.com/forum.php?mod=viewthread&tid=7704
(出处: about云开发)


Hadoop的初学者经常会疑惑这样两个问题:
1.Hadoop的一个Block默认是64M,那么对于一个记录行形式的文本,会不会造成一行记录被分到两个Block当中?
2.在把文件从Block中读取出来进行切分时,会不会造成一行记录被分成两个InputSplit,如果被分成两个InputSplit,这样一个InputSplit里面就有一行不完整的数据,那么处理这个InputSplit的Mapper会不会得出不正确的结果?
对于上面的两个问题,首先要明确两个概念:Block和InputSplit
1. block是hdfs存储文件的单位(默认是64M);
2. InputSplit是MapReduce对文件进行处理和运算的输入单位,只是一个逻辑概念,每个InputSplit并没有对文件实际的切割,只是记录了要处理的数据的位置(包括文件的path和hosts)和长度(由start和length决定)。

因此,以行记录形式的文本,还真可能存在一行记录被划分到不同的Block,甚至不同的DataNode上去。通过分析FileInputFormat里面的getSplits方法,可以得出,某一行记录同样也可能被划分到不同的InputSplit。

public List<InputSplit> getSplits(JobContext job) throws IOException {  long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  long maxSize = getMaxSplitSize(job);  // generate splits  List<InputSplit> splits = new ArrayList<InputSplit>();  List<FileStatus> files = listStatus(job);        for (FileStatus file: files) {  Path path = file.getPath();  long length = file.getLen();  if (length != 0) {  FileSystem fs = path.getFileSystem(job.getConfiguration());  BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);  if (isSplitable(job, path)) {  long blockSize = file.getBlockSize();  long splitSize = computeSplitSize(blockSize, minSize, maxSize);  long bytesRemaining = length;  while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);  splits.add(makeSplit(path, length-bytesRemaining, splitSize,  blkLocations[blkIndex].getHosts()));  bytesRemaining -= splitSize;  }  if (bytesRemaining != 0) {  splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,  blkLocations[blkLocations.length-1].getHosts()));  }  } else { // not splitable  splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));  }  } else {   //Create empty hosts array for zero length files  splits.add(makeSplit(path, 0, length, new String[0]));  }  }  // Save the number of input files for metrics/loadgen  job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());  LOG.debug("Total # of splits: " + splits.size());  return splits;  
}  

复制代码

从上面的代码可以看出,对文件进行切分其实很简单:获取文件在HDFS上的路径和Block信息,然后根据splitSize
对文件进行切分,splitSize = computeSplitSize(blockSize, minSize, maxSize);blockSize,minSize,maxSize都可以配置,默认splitSize 就等于blockSize的默认值(64m)。
FileInputFormat对文件的切分是严格按照偏移量来的,因此一行记录比较长的话,其可能被切分到不同的InputSplit。但这并不会对Map造成影响,尽管一行记录可能被拆分到不同的InputSplit,但是与FileInputFormat关联的RecordReader被设计的足够健壮,当一行记录跨InputSplit时,其能够到读取不同的InputSplit,直到把这一行记录读取完成,在Hadoop里,记录行形式的文本,通常采用默认的TextInputFormat,TextInputFormat关联的是LineRecordReader,下面我们来看看LineRecordReader的的nextKeyValue方法里读取文件的代码:

while (getFilePosition() <= end) {  newSize = in.readLine(value, maxLineLength,  Math.max(maxBytesToConsume(pos), maxLineLength));  if (newSize == 0) {  break;  }  

复制代码
其读取文件是通过LineReader(in就是一个LineReader实例)的readLine方法完成的:

public int readLine(Text str, int maxLineLength,  int maxBytesToConsume) throws IOException {  if (this.recordDelimiterBytes != null) {  return readCustomLine(str, maxLineLength, maxBytesToConsume);  } else {  return readDefaultLine(str, maxLineLength, maxBytesToConsume);  }  
}  /** * Read a line terminated by one of CR, LF, or CRLF. */  
private int readDefaultLine(Text str, int maxLineLength, int maxBytesToConsume)  
throws IOException {  str.clear();  int txtLength = 0; //tracks str.getLength(), as an optimization  int newlineLength = 0; //length of terminating newline  boolean prevCharCR = false; //true of prev char was CR  long bytesConsumed = 0;  do {  int startPosn = bufferPosn; //starting from where we left off the last time  if (bufferPosn >= bufferLength) {  startPosn = bufferPosn = 0;  if (prevCharCR)  ++bytesConsumed; //account for CR from previous read  bufferLength = in.read(buffer);  if (bufferLength <= 0)  break; // EOF  }  for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline  if (buffer[bufferPosn] == LF) {  newlineLength = (prevCharCR) ? 2 : 1;  ++bufferPosn; // at next invocation proceed from following byte  break;  }  if (prevCharCR) { //CR + notLF, we are at notLF  newlineLength = 1;  break;  }  prevCharCR = (buffer[bufferPosn] == CR);  }  int readLength = bufferPosn - startPosn;  if (prevCharCR && newlineLength == 0)  --readLength; //CR at the end of the buffer  bytesConsumed += readLength;  int appendLength = readLength - newlineLength;  if (appendLength > maxLineLength - txtLength) {  appendLength = maxLineLength - txtLength;  }  if (appendLength > 0) {  str.append(buffer, startPosn, appendLength);  txtLength += appendLength;  }  } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);   <span style="color: #ff0000;">//①</span>  if (bytesConsumed > (long)Integer.MAX_VALUE)  throw new IOException("Too many bytes before newline: " + bytesConsumed);      return (int)bytesConsumed;  
}  

复制代码

我们分析下readDefaultLine方法,do-while循环体主要是读取文件,然后遍历读取的内容,找到默认的换行符就终止循环。前面说,对于跨InputSplit的行,LineRecordReader会自动跨InputSplit去读取。这就体现在上述代码的While循环的终止条件上:
while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);
newlineLength==0则以为一次do-while循环中读取的内容中没有遇到换行符,因maxBytesToConsume的默认值为Integer.MAX_VALUE,所以如果读取的内容没有遇到换行符,则会一直读取下去,知道读取的内容超过maxBytesToConsume。这样的出来方式,解决了一行记录跨InputSplit的读取问题,同样也会造成下面两个疑问:
1.既然在LineReader读取方法里面没有对考虑InputSplit的end进行处理,难道读取一个InputSplit的时候,会这样无限的读取下去么?
2.如果一行记录L跨越了A,B两个InputSplit,读A的时候已经读取了跨越A,B的这条记录L,那么对B这个InputSplit读取的时候,如果做到不读取L这条记录在B中的部分呢?
为了解决这两个问题,Hadoop通过下面的代码来做到: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;  // We always read one extra line, which lies outside the upper  // split limit i.e. (end - 1)  while (getFilePosition() <= end) {        <span style="color: #ff0000;"> //②</span>  newSize = in.readLine(value, maxLineLength,  Math.max(maxBytesToConsume(pos), maxLineLength));  if (newSize == 0) {  break;  }  pos += newSize;  inputByteCounter.increment(newSize);  if (newSize < maxLineLength) {  break;  }  // line too long. try again  LOG.info("Skipped line of size " + newSize + " at pos " +   (pos - newSize));  }  if (newSize == 0) {  key = null;  value = null;  return false;  } else {  return true;  }  
}  

复制代码

过代码②处得While条件,就保证了InputSplit读取边界的问题,如果存在跨InputSplit的记录,也只好跨InputSplit读取一次。
再来看LineRecordReader的initialize方法:

// If this is not the first split, we always throw away first record  
// because we always (except the last split) read one extra line in  
// next() method.  
if (start != 0) {  start += in.readLine(new Text(), 0, maxBytesToConsume(start));  
}  
this.pos = start;  

复制代码
如果不是第一InputSplit,则在读取的时候,LineRecordReader会自动忽略掉第一个换行符之前的所有内容,这样就不存在重读读取的问题。

个人理解

这篇关于Hadoop MapReduce中如何处理跨行Block和inputSplit的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1104847

相关文章

Springboot处理跨域的实现方式(附Demo)

《Springboot处理跨域的实现方式(附Demo)》:本文主要介绍Springboot处理跨域的实现方式(附Demo),具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不... 目录Springboot处理跨域的方式1. 基本知识2. @CrossOrigin3. 全局跨域设置4.

python+opencv处理颜色之将目标颜色转换实例代码

《python+opencv处理颜色之将目标颜色转换实例代码》OpenCV是一个的跨平台计算机视觉库,可以运行在Linux、Windows和MacOS操作系统上,:本文主要介绍python+ope... 目录下面是代码+ 效果 + 解释转HSV: 关于颜色总是要转HSV的掩膜再标注总结 目标:将红色的部分滤

Python实现自动化接收与处理手机验证码

《Python实现自动化接收与处理手机验证码》在移动互联网时代,短信验证码已成为身份验证、账号注册等环节的重要安全手段,本文将介绍如何利用Python实现验证码的自动接收,识别与转发,需要的可以参考下... 目录引言一、准备工作1.1 硬件与软件需求1.2 环境配置二、核心功能实现2.1 短信监听与获取2.

Python使用date模块进行日期处理的终极指南

《Python使用date模块进行日期处理的终极指南》在处理与时间相关的数据时,Python的date模块是开发者最趁手的工具之一,本文将用通俗的语言,结合真实案例,带您掌握date模块的六大核心功能... 目录引言一、date模块的核心功能1.1 日期表示1.2 日期计算1.3 日期比较二、六大常用方法详

利用Go语言开发文件操作工具轻松处理所有文件

《利用Go语言开发文件操作工具轻松处理所有文件》在后端开发中,文件操作是一个非常常见但又容易出错的场景,本文小编要向大家介绍一个强大的Go语言文件操作工具库,它能帮你轻松处理各种文件操作场景... 目录为什么需要这个工具?核心功能详解1. 文件/目录存javascript在性检查2. 批量创建目录3. 文件

Java使用多线程处理未知任务数的方案介绍

《Java使用多线程处理未知任务数的方案介绍》这篇文章主要为大家详细介绍了Java如何使用多线程实现处理未知任务数,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 知道任务个数,你可以定义好线程数规则,生成线程数去跑代码说明:1.虚拟线程池:使用 Executors.newVir

一文带你深入了解Python中的GeneratorExit异常处理

《一文带你深入了解Python中的GeneratorExit异常处理》GeneratorExit是Python内置的异常,当生成器或协程被强制关闭时,Python解释器会向其发送这个异常,下面我们来看... 目录GeneratorExit:协程世界的死亡通知书什么是GeneratorExit实际中的问题案例

最新Spring Security实战教程之表单登录定制到处理逻辑的深度改造(最新推荐)

《最新SpringSecurity实战教程之表单登录定制到处理逻辑的深度改造(最新推荐)》本章节介绍了如何通过SpringSecurity实现从配置自定义登录页面、表单登录处理逻辑的配置,并简单模拟... 目录前言改造准备开始登录页改造自定义用户名密码登陆成功失败跳转问题自定义登出前后端分离适配方案结语前言

Gin框架中的GET和POST表单处理的实现

《Gin框架中的GET和POST表单处理的实现》Gin框架提供了简单而强大的机制来处理GET和POST表单提交的数据,通过c.Query、c.PostForm、c.Bind和c.Request.For... 目录一、GET表单处理二、POST表单处理1. 使用c.PostForm获取表单字段:2. 绑定到结

mysql8.0无备份通过idb文件恢复数据的方法、idb文件修复和tablespace id不一致处理

《mysql8.0无备份通过idb文件恢复数据的方法、idb文件修复和tablespaceid不一致处理》文章描述了公司服务器断电后数据库故障的过程,作者通过查看错误日志、重新初始化数据目录、恢复备... 周末突然接到一位一年多没联系的妹妹打来电话,“刘哥,快来救救我”,我脑海瞬间冒出妙瓦底,电信火苲马扁.