课程表

Hadoop课程

工具箱
速查手册

Hadoop MapReduce操作

当前位置：免费教程 » 大数据/云 » Hadoop

MapReduce - 读取数据

通过InputFormat决定读取的数据的类型，然后拆分成一个个InputSplit，每个InputSplit对应一个Map处理，RecordReader读取InputSplit的内容给Map

InputFormat

决定读取数据的格式，可以是文件或数据库等

功能

  1. 验证作业输入的正确性，如格式等
  2. 将输入文件切割成逻辑分片(InputSplit)，一个InputSplit将会被分配给一个独立的Map任务
  3. 提供RecordReader实现，读取InputSplit中的"K-V对"供Mapper使用

方法

List getSplits(): 获取由输入文件计算出输入分片(InputSplit)，解决数据或文件分割成片问题

RecordReader <k,v>createRecordReader():</k,v> 创建RecordReader，从InputSplit中读取数据，解决读取分片中数据问题

类结构

TextInputFormat: 输入文件中的每一行就是一个记录，Key是这一行的byte offset，而value是这一行的内容

KeyValueTextInputFormat: 输入文件中每一行就是一个记录，第一个分隔符字符切分每行。在分隔符字符之前的内容为Key，在之后的为Value。分隔符变量通过key.value.separator.in.input.line变量设置，默认为(\t)字符。

NLineInputFormat: 与TextInputFormat一样，但每个数据块必须保证有且只有Ｎ行，mapred.line.input.format.linespermap属性，默认为１

SequenceFileInputFormat: 一个用来读取字符流数据的InputFormat，<key,value>为用户自定义的。字符流数据是Hadoop自定义的压缩的二进制数据格式。它用来优化从一个MapReduce任务的输出到另一个MapReduce任务的输入之间的数据传输过程。</key,value>

InputSplit

代表一个个逻辑分片，并没有真正存储数据，只是提供了一个如何将数据分片的方法

Split内有Location信息，利于数据局部化

一个InputSplit给一个单独的Map处理

  1. public abstract class InputSplit {
  2.       /**
  3.        * 获取Split的大小，支持根据size对InputSplit排序.
  4.        */
  5.       public abstract long getLength() throws IOException, InterruptedException;
  7.       /**
  8.        * 获取存储该分片的数据所在的节点位置.
  9.        */
  10.       public abstract String[] getLocations() throws IOException, InterruptedException;
  11. }

RecordReader

将InputSplit拆分成一个个<key,value>对给Map处理，也是实际的文件读取分隔对象</key,value>

问题

大量小文件如何处理

CombineFileInputFormat可以将若干个Split打包成一个，目的是避免过多的Map任务（因为Split的数目决定了Map的数目，大量的Mapper Task创建销毁开销将是巨大的）

怎么计算split的

通常一个split就是一个block（FileInputFormat仅仅拆分比block大的文件），这样做的好处是使得Map可以在存储有当前数据的节点上运行本地的任务，而不需要通过网络进行跨节点的任务调度

通过mapred.min.split.size， mapred.max.split.size, block.size来控制拆分的大小

如果mapred.min.split.size大于block size，则会将两个block合成到一个split，这样有部分block数据需要通过网络读取

如果mapred.max.split.size小于block size，则会将一个block拆成多个split，增加了Map任务数（Map对split进行计算并且上报结果，关闭当前计算打开新的split均需要耗费资源）

先获取文件在HDFS上的路径和Block信息，然后根据splitSize对文件进行切分（ splitSize = computeSplitSize(blockSize, minSize, maxSize) ），默认splitSize 就等于blockSize的默认值（64m）

  1. public List<InputSplit> getSplits(JobContext job) throws IOException {
  2.     // 首先计算分片的最大和最小值。这两个值将会用来计算分片的大小
  3.     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
  4.     long maxSize = getMaxSplitSize(job);
  6.     // generate splits
  7.     List<InputSplit> splits = new ArrayList<InputSplit>();
  8.     List<FileStatus> files = listStatus(job);
  9.     for (FileStatus file: files) {
  10.         Path path = file.getPath();
  11.         long length = file.getLen();
  12.         if (length != 0) {
  13.               FileSystem fs = path.getFileSystem(job.getConfiguration());
  14.             // 获取该文件所有的block信息列表[hostname, offset, length]
  15.               BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
  16.             // 判断文件是否可分割，通常是可分割的，但如果文件是压缩的，将不可分割
  17.               if (isSplitable(job, path)) {
  18.                 long blockSize = file.getBlockSize();
  19.                 // 计算分片大小
  20.                 // 即 Math.max(minSize, Math.min(maxSize, blockSize));
  21.                 long splitSize = computeSplitSize(blockSize, minSize, maxSize);
  23.                 long bytesRemaining = length;
  24.                 // 循环分片。
  25.                 // 当剩余数据与分片大小比值大于Split_Slop时，继续分片， 小于等于时，停止分片
  26.                 while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  27.                       int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  28.                       splits.add(makeSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));
  29.                       bytesRemaining -= splitSize;
  30.                 }
  31.                 // 处理余下的数据
  32.                 if (bytesRemaining != 0) {
  33.                     splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
  34.                 }
  35.             } else {
  36.                 // 不可split，整块返回
  37.                 splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
  38.             }
  39.         } else {
  40.             // 对于长度为0的文件，创建空Hosts列表，返回
  41.             splits.add(makeSplit(path, 0, length, new String[0]));
  42.         }
  43.     }
  45.     // 设置输入文件数量
  46.     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
  47.     LOG.debug("Total # of splits: " + splits.size());
  48.     return splits;
  49. }

分片间的数据如何处理

split是根据文件大小分割的，而一般处理是根据分隔符进行分割的，这样势必存在一条记录横跨两个split

解决办法是只要不是第一个split，都会远程读取一条记录。不是第一个split的都忽略到第一条记录

  1. public class LineRecordReader extends RecordReader<LongWritable, Text> {
  2.     private CompressionCodecFactory compressionCodecs = null;
  3.     private long start;
  4.     private long pos;
  5.     private long end;
  6.     private LineReader in;
  7.     private int maxLineLength;
  8.     private LongWritable key = null;
  9.     private Text value = null;
  11.     // initialize函数即对LineRecordReader的一个初始化
  12.     // 主要是计算分片的始末位置，打开输入流以供读取K-V对，处理分片经过压缩的情况等
  13.     public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException {
  14.         FileSplit split = (FileSplit) genericSplit;
  15.         Configuration job = context.getConfiguration();
  16.         this.maxLineLength = job.getInt("mapred.linerecordreader.maxlength", Integer.MAX_VALUE);
  17.         start = split.getStart();
  18.         end = start + split.getLength();
  19.         final Path file = split.getPath();
  20.         compressionCodecs = new CompressionCodecFactory(job);
  21.         final CompressionCodec codec = compressionCodecs.getCodec(file);
  23.         // 打开文件，并定位到分片读取的起始位置
  24.         FileSystem fs = file.getFileSystem(job);
  25.         FSDataInputStream fileIn = fs.open(split.getPath());
  27.         boolean skipFirstLine = false;
  28.         if (codec != null) {
  29.             // 文件是压缩文件的话，直接打开文件
  30.             in = new LineReader(codec.createInputStream(fileIn), job);
  31.             end = Long.MAX_VALUE;
  32.         } else {
  33.             // 只要不是第一个split，则忽略本split的第一行数据
  34.             if (start != 0) {
  35.                 skipFirstLine = true;
  36.                 --start;
  37.                 // 定位到偏移位置，下次读取就会从偏移位置开始
  38.                 fileIn.seek(start);
  39.             }
  40.             in = new LineReader(fileIn, job);
  41.         }
  43.         if (skipFirstLine) {
  44.             // 忽略第一行数据，重新定位start
  45.             start += in.readLine(new Text(), 0, (int) Math.min((long) Integer.MAX_VALUE, end - start));
  46.         }
  47.         this.pos = start;
  48.     }
  50.     public boolean nextKeyValue() throws IOException {
  51.         if (key == null) {
  52.             key = new LongWritable();
  53.         }
  54.         key.set(pos);// key即为偏移量
  55.         if (value == null) {
  56.             value = new Text();
  57.         }
  58.         int newSize = 0;
  59.         while (pos < end) {
  60.             newSize = in.readLine(value, maxLineLength,    Math.max((int) Math.min(Integer.MAX_VALUE, end - pos), maxLineLength));
  61.             // 读取的数据长度为0，则说明已读完
  62.             if (newSize == 0) {
  63.                 break;
  64.             }
  65.             pos += newSize;
  66.             // 读取的数据长度小于最大行长度，也说明已读取完毕
  67.             if (newSize < maxLineLength) {
  68.                 break;
  69.             }
  70.             // 执行到此处，说明该行数据没读完，继续读入
  71.         }
  72.         if (newSize == 0) {
  73.             key = null;
  74.             value = null;
  75.             return false;
  76.         } else {
  77.             return true;
  78.         }
  79.     }
  80. }

MapReduce - Mapper

主要是读取InputSplit的每一个Key,Value对并进行处理

  1. public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
  2.     /**
  3.      * 预处理，仅在map task启动时运行一次
  4.      */
  5.     protected void setup(Context context) throws  IOException, InterruptedException {
  6.     }
  8.     /**
  9.      * 对于InputSplit中的每一对<key, value>都会运行一次
  10.      */
  11.     @SuppressWarnings("unchecked")
  12.     protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException {
  13.         context.write((KEYOUT) key, (VALUEOUT) value);
  14.     }
  16.     /**
  17.      * 扫尾工作，比如关闭流等
  18.      */
  19.     protected void cleanup(Context context) throws IOException, InterruptedException {
  20.     }
  22.     /**
  23.      * map task的驱动器
  24.      */
  25.     public void run(Context context) throws IOException, InterruptedException {
  26.         setup(context);
  27.         while (context.nextKeyValue()) {
  28.             map(context.getCurrentKey(), context.getCurrentValue(), context);
  29.         }
  30.         cleanup(context);
  31.     }
  32. }
  34. public class MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskInputOutputContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
  35.     private RecordReader<KEYIN, VALUEIN> reader;
  36.     private InputSplit split;
  38.     /**
  39.      * Get the input split for this map.
  40.      */
  41.     public InputSplit getInputSplit() {
  42.         return split;
  43.     }
  45.     @Override
  46.     public KEYIN getCurrentKey() throws IOException, InterruptedException {
  47.         return reader.getCurrentKey();
  48.     }
  50.     @Override
  51.     public VALUEIN getCurrentValue() throws IOException, InterruptedException {
  52.         return reader.getCurrentValue();
  53.     }
  55.     @Override
  56.     public boolean nextKeyValue() throws IOException, InterruptedException {
  57.         return reader.nextKeyValue();
  58.     }
  59. }

MapReduce - Shuffle

对Map的结果进行排序并传输到Reduce进行处理 Map的结果并不是直接存放到硬盘,而是利用缓存做一些预排序处理 Map会调用Combiner，压缩，按key进行分区、排序等，尽量减少结果的大小 每个Map完成后都会通知Task，然后Reduce就可以进行处理

Map端

当Map程序开始产生结果的时候，并不是直接写到文件的，而是利用缓存做一些排序方面的预处理操作

每个Map任务都有一个循环内存缓冲区（默认100MB），当缓存的内容达到80%时，后台线程开始将内容写到文件，此时Map任务可以继续输出结果，但如果缓冲区满了，Map任务则需要等待

写文件使用round-robin方式。在写入文件之前，先将数据按照Reduce进行分区。对于每一个分区，都会在内存中根据key进行排序，如果配置了Combiner，则排序后执行Combiner（Combine之后可以减少写入文件和传输的数据）

每次结果达到缓冲区的阀值时，都会创建一个文件，在Map结束时，可能会产生大量的文件。在Map完成前，会将这些文件进行合并和排序。如果文件的数量超过3个，则合并后会再次运行Combiner（1、2个文件就没有必要了）

如果配置了压缩，则最终写入的文件会先进行压缩，这样可以减少写入和传输的数据

一旦Map完成，则通知任务管理器，此时Reduce就可以开始复制结果数据

Reduce端

Map的结果文件都存放到运行Map任务的机器的本地硬盘中

如果Map的结果很少，则直接放到内存，否则写入文件中

同时后台线程将这些文件进行合并和排序到一个更大的文件中（如果文件是压缩的，则需要先解压）

当所有的Map结果都被复制和合并后，就会调用Reduce方法

Reduce结果会写入到HDFS中

调优

一般的原则是给shuffle分配尽可能多的内存，但前提是要保证Map、Reduce任务有足够的内存

对于Map，主要就是避免把文件写入磁盘，例如使用Combiner，增大io.sort.mb的值

对于Reduce，主要是把Map的结果尽可能地保存到内存中，同样也是要避免把中间结果写入磁盘。默认情况下，所有的内存都是分配给Reduce方法的，如果Reduce方法不怎么消耗内存，可以mapred.inmem.merge.threshold设成0，mapred.job.reduce.input.buffer.percent设成1.0

在任务监控中可通过Spilled records counter来监控写入磁盘的数，但这个值是包括map和reduce的

对于IO方面，可以Map的结果可以使用压缩，同时增大buffer size（io.file.buffer.size，默认4kb）

配置

属性 默认值 描述
io.sort.mb 100 映射输出分类时所使用缓冲区的大小.
io.sort.record.percent 0.05 剩余空间用于映射输出自身记录.在1.X发布后去除此属性.随机代码用于使用映射所有内存并记录信息.
io.sort.spill.percent 0.80 针对映射输出内存缓冲和记录索引的阈值使用比例.
io.sort.factor 10 文件分类时合并流的最大数量。此属性也用于reduce。通常把数字设为100.
min.num.spills.for.combine 3 组合运行所需最小溢出文件数目.
mapred.compress.map.output false 压缩映射输出.
mapred.map.output.compression.codec DefaultCodec 映射输出所需的压缩解编码器.
mapred.reduce.parallel.copies 5 用于向reducer传送映射输出的线程数目.
mapred.reduce.copy.backoff 300 时间的最大数量，以秒为单位，这段时间内若reducer失败则会反复尝试传输
io.sort.factor 10 组合运行所需最大溢出文件数目.
mapred.job.shuffle.input.buffer.percent 0.70 随机复制阶段映射输出缓冲器的堆栈大小比例
mapred.job.shuffle.merge.percent 0.66 用于启动合并输出进程和磁盘传输的映射输出缓冲器的阀值使用比例
mapred.inmem.merge.threshold 1000 用于启动合并输出和磁盘传输进程的映射输出的阀值数目。小于等于0意味着没有门槛，而溢出行为由 mapred.job.shuffle.merge.percent单独管理.
mapred.job.reduce.input.buffer.percent 0.0 用于减少内存映射输出的堆栈大小比例，内存中映射大小不得超出此值。若reducer需要较少内存则可以提高该值.

MapReduce - 编程

在线练习

http://cloudcomputing.ruc.edu.cn

处理

  1. select：直接分析输入数据，取出需要的字段数据即可
  2. where: 也是对输入数据处理的过程中进行处理，判断是否需要该数据
  3. aggregation:min, max, sum
  4. group by: 通过Reducer实现
  5. sort
  6. join: map join, reduce join

Third-Party Libraries

export LIBJARS=$MYLIB/commons-lang-2.3.jar, hadoop jar prohadoop-0.0.1-SNAPSHOT.jar org.aspress.prohadoop.c3. WordCountUsingToolRunner -libjars $LIBJARS

hadoop jar prohadoop-0.0.1-SNAPSHOT-jar-with-dependencies.jar org.aspress.prohadoop.c3. WordCountUsingToolRunner The dependent libraries are now included inside the application JAR file

一般还是上面的好，指定依赖可以利用Public Cache，如果是包含依赖，则每次都需要拷贝

参考书籍

MapReduce Design Patterns

转载本站内容时，请务必注明来自W3xue，违者必究。
 友情链接：直通硅谷 　点职佳 　北美留学生论坛

本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
皖ICP备17017327号-2　皖公网安备34020702000426号