Hadoop处理记录如何跨块边界拆分？

Question

根据Hadoop--权威指南

FileInputFormats定义的逻辑记录通常不适合HDFS块。例如，TextInputFormat的逻辑记录是行，它们通常会跨越HDFS边界。这与程序的功能无关--例如，行不会丢失或中断--但值得了解，因为这确实意味着数据本地映射（即与输入数据运行在同一主机上的映射）将执行一些远程读取。这导致的轻微开销通常并不显著。

假设记录行被分成两个块(b1和b2)。处理第一个块(b1)的映射器将注意到最后一行没有EOL分隔符，并从下一个数据块(b2)中获取该行的剩余部分。

处理第二块(b2)的映射器如何确定第一记录是不完整的，并且应该从块(b2)中的第二记录开始处理？

Answer 1

有趣的问题，我花了一些时间查看代码的细节，下面是我的想法。客户机通过inputformat.getsplits处理拆分，因此查看FileInputFormat可以得到以下信息：

>

对于每个输入文件，获取文件长度和块大小，并将拆分大小计算为max(minSize，min(maxSize，blockSize))，其中maxSize对应于mapred.max.split.size，minSizemapred.min.split.size.

根据上面计算的拆分大小将文件划分为不同的filesplit。这里重要的是，每个filesplit都用一个start参数初始化，该参数对应于输入文件中的偏移量。在那一点上仍然没有对线路的处理。代码的相关部分如下所示：

while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
                           blkLocations[blkIndex].getHosts()));
  bytesRemaining -= splitSize;
}

之后，如果查看由TextInputFormat定义的LineRecordReader，那么行就是在这里处理的：

>

初始化LineRecordReader时，它试图实例化LineReader，这是一种抽象，可以读取FSDataInputStream上的行。有2种情况:

如果定义了compressioncodec，那么这个编解码器负责处理边界。可能与您的问题无关。

但是，如果没有编解码器，这就是有趣的地方：如果inputsplit的start不同于0，那么您将回溯1个字符，然后跳过您遇到的由\n或\r\n（Windows）标识的第一行！回溯很重要，因为如果您的行边界与拆分边界相同，这将确保您不会跳过有效的行。以下是相关代码：

if (codec != null) {
   in = new LineReader(codec.createInputStream(fileIn), job);
   end = Long.MAX_VALUE;
} else {
   if (start != 0) {
     skipFirstLine = true;
     --start;
     fileIn.seek(start);
   }
   in = new LineReader(fileIn, job);
}
if (skipFirstLine) {  // skip first line and re-establish "start".
  start += in.readLine(new Text(), 0,
                    (int)Math.min((long)Integer.MAX_VALUE, end - start));
}
this.pos = start;

因此，由于拆分是在客户机中计算的，映射器不需要按顺序运行，每个映射器都已经知道是否需要丢弃第一行。

所以基本上，如果在同一个文件中有两行，每行100MB，为了简化，假设拆分大小为64MB。然后当计算输入分裂时，我们将有以下场景：

• 包含此块的路径和主机的拆分1。开始时初始化200-200=0MB，长度64MB.
• 拆分2在开始时初始化200-200+64=64MB，长度64MB.
• 拆分3在开始时初始化200-200+128=128MB，长度64MB.
• 拆分4在开始时初始化200-200+192=192MB，长度8MB.
• 映射器A将处理拆分1，开始是0，所以不要跳过第一行，而是读取超过64MB限制的整行，所以需要远程读取。
• 映射器B将处理split 2,start is！=0所以跳过64MB-1Byte之后的第一行，它对应于第1行的100MB结尾，仍然在split 2中，我们在split 2中有28MB的行，所以远程读取剩余的72MB。
• 映射器C将处理split 3,start is！=0所以跳过128MB-1byte之后的第一行，它对应于200MB的第2行的结尾，这是文件的结尾，所以不要做任何事情。
• 映射器D与映射器C相同，只是它在192MB-1字节之后查找换行符。