Java大数据开发Hadoop MapReduce-662P.COM

MapReduce的思想核心是“分而治之”，适用于大量复杂的任务处理场景（大规模数据处理场景）Map负责“分”，即把复杂的任务分解为若干个“简单的任务”来并行处理。可以进行拆分的前提是这些小任务可以并行计算，彼此间几乎没有依赖关系

1 MapRedcue的介绍

1.1 MapReduce定义

MapReduce是一个进行分布式运算的编程框架，使用户开发基于hadoop进行数据分析的核心框架。 MapReduce 核心功能就是将用户编写的业务逻辑代码和自带的默认组件整合成一个完整的分布式运算程序，并发运行在一个 Hadoop 集群上。

1.2 MapReduce的思想

MapReduce的思想核心是分而治之，适用于大规模数据处理场景。

map负责分，将复杂的任务拆解成可以并行计算的若干个任务来处理

reduce负责合，对map阶段的结果进行全局汇总

比如说：老师作业留的有点多，一个人写太费劲了，就可以用MapReduce这种分而治之的思想，将作业进行map处理，分给不同的人，最后所有写完的部分发到群里进行reduce汇总，复杂的作业简简单单。

1.3 MapReduce优点

易于编程

MapReduce将做什么和怎么做分开了，提供了一些接口，程序员只需关注应用层上的问题。具体如何实现并行计算任务则被隐藏了起来。

扩展性

当计算资源不足时，可以增加机器来提高扩展能力

高容错

一台机器挂了，可以将计算任务转移到另一台节点上进行

适合PB级海量数据的离线处理

1.4 MapReduce的缺点

不擅长实时计算

无法做到在毫秒级别返回结果

不擅长流式计算

MapReduce处理的数据源只能是静态的，不能动态变化

不擅长DAG(有向无环图)计算

每个MR作业处理结束，结果都会写入到磁盘，造成大量的磁盘IO，导致性能低下

1.5 MapReduce进程

一个MapReduce程序在分布式运行时有三类的实例进程

MrAppMaster : 负责整个程序的过程调度及状态协调
MapTask : 负责Map阶段的数据处理流程
ReduceTask ：负责Reduce阶段的数据处理流程

1.6 MapReduce-WordCount

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class WordCount {
  public static class TokenizerMapper
       extends Mapper<Object, Text, Text, IntWritable>{
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    public void map(Object key, Text value, Context context) 
    throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }
  public static class IntSumReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable();
    public void reduce(Text key, Iterable<IntWritable> 
    values,Context context) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }
  }
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}

以上代码实现了两个类：TokenizerMapper和IntSumReducer，它们分别实现了Map和Reduce功能。

Map函数将输入的每一行文本进行分词，并将每个单词映射为一个键值对，其中键为单词，值为1，然后将这些键值对输出给Reduce函数。

Reduce函数将相同键的值相加，并将最终结果输出。

在这个例子中，CombinerClass被设置为相同的Reducer类，用于在Map任务结束后本地合并中间结果，以减少网络传输。

最后，将输入文件和输出文件的路径作为命令行参数传递给main函数，并启动MapReduce作业。

job的讲解

在Hadoop MapReduce程序中，Job对象是用来定义和运行一个MapReduce作业的。

Job对象的主要功能是封装了整个MapReduce作业的配置和运行信息，包括输入数据和输出数据的路径、Mapper类和Reducer类的设置、中间结果的输出类型和格式、作业的提交方式等。

在main函数中，我们创建一个Job对象并设置它的相关属性。

Job.getInstance()方法返回一个新的Job实例，其中的Configuration对象用来指定作业的一些配置信息。

setJarByClass()方法用来设置作业的jar包，它的参数是定义MapReduce作业的主类。

setMapperClass()、setCombinerClass()和setReducerClass()方法用来指定Mapper、Combiner和Reducer的实现类。

setOutputKeyClass()和setOutputValueClass()方法分别用来设置MapReduce作业的输出键和输出值的类型。

FileInputFormat.addInputPath()和FileOutputFormat.setOutputPath()方法用来指定输入文件和输出文件的路径。

最后，我们调用job.waitForCompletion()方法来提交并运行作业，并等待作业完成。

如果作业成功完成，waitForCompletion()方法将返回true，否则返回false。

如果作业失败，我们可以通过job.getJobState()方法来获取作业的状态信息，或者查看作业的日志信息来进行排错和调试。

2 Hadoop序列化

2.1 序列化的定义

序列化就是将内存中对象转换成字节序列，便于存储到磁盘和网络传输

反序列化时将字节序列或磁盘中的持久化数据转换成内存中的对象

一般来说，对象只能在本地进程中使用，不能通过网络发送到另一台计算机

序列化可以存储对象，可以将对象发送到远程计算机

2.2 hadoop序列化和java序列化的区别

Hadoop序列化和Java序列化都是将对象转换为字节序列以便于在网络上传输或者存储到磁盘等持久化存储介质中。它们的主要区别在于以下几点：

序列化速度和效率不同：Hadoop序列化比Java序列化更快，因为它采用的是二进制格式，而Java序列化采用的是基于文本的XML或JSON格式。
支持的数据类型不同：Hadoop序列化支持的数据类型比Java序列化更多，包括基本类型、数组、集合、映射、枚举、自定义类等。
序列化后的数据大小不同：Hadoop序列化生成的字节流比Java序列化生成的字节流更小，因为它使用更紧凑的二进制格式，这对于在网络上传输和存储到磁盘等介质中非常重要。
可移植性不同：Java序列化生成的字节流只能被Java程序读取，而Hadoop序列化生成的字节流可以被任何语言的程序读取，因为它使用了通用的二进制格式。

总的来说，Hadoop序列化更适合用于大规模数据的处理和分布式计算，而Java序列化更适合用于小规模数据的传输和存储。

3 MapReduce 的原理

3.1 MapReduce 工作的过程

数据输入：MapReduce从Hadoop分布式文件系统（HDFS）中读取输入数据，并将其分成固定大小的数据块，每个数据块大小通常为64MB。

Map阶段：在Map阶段，MapReduce将每个数据块分发给一组可扩展的计算节点，每个计算节点运行Map函数来处理它们分配的数据块。Map函数将输入数据转换为一组键值对（Key-Value Pairs）的形式，这些键值对可以被后续的Reduce函数处理。

Shuffle阶段：在Map函数处理完数据之后，MapReduce框架将所有的键值对按照它们的键进行排序，并将相同键的值合并在一起。这个过程通常被称为“Shuffle”。

Reduce阶段：在Reduce阶段，MapReduce框架将合并后的键值对发送到一组可扩展的计算节点。每个节点运行Reduce函数来处理它们收到的所有键值对，并生成最终的输出结果。

数据输出：在Reduce函数处理完数据之后，MapReduce将输出结果写入HDFS中。

这些步骤中，Map和Reduce函数是由开发者自行编写的，它们实现了具体的业务逻辑。MapReduce框架提供了分布式计算的基础设施，负责管理计算节点、任务分配、故障处理等任务，以保证整个计算过程的可靠性和高效性。

总的来说，MapReduce框架的原理是将大数据集划分成多个小数据块，然后将这些数据块分发给多个计算节点并行处理，最后将处理结果合并为一个最终结果。它通过这种方式来充分利用集群中的计算资源，提高计算效率和数据处理能力。

3.2 InputFormat 数据输入

3.2.1 切片

数据块：Block时HDFS在物理上对数据进行切块，是HDFS存储数据的单位

数据切片：数据切片是在逻辑上对输入进行切片。切片是MR程序计算输入数据的单位，一个切片会启动一个MapTask

客户端提交job时的切片数决定了map阶段的并行度

默认情况下，切片大小为BlockSize

切片不会考虑数据整体，是对每个文件进行单独切片

3.2.2 FileInputFormat

在MapReduce中，FileInputFormat是一个抽象类，用于定义如何将文件分割成输入数据块并生成适合Mapper处理的RecordReader。它是MapReduce中的输入格式类之一，用于读取Hadoop分布式文件系统（HDFS）或本地文件系统中的数据。

FileInputFormat包括两个关键方法：getSplits()和createRecordReader()。

getSplits()方法将输入文件划分成适合Map任务的数据块，每个数据块对应一个Map任务。该方法返回一个InputSplit对象的数组，其中每个InputSplit表示一个文件数据块。

createRecordReader()方法创建一个RecordReader对象，用于读取InputSplit中的数据块。RecordReader负责读取一个数据块中的所有记录，并将它们转换成key-value对。

FileInputFormat还提供了一些其他的方法，如isSplitable()用于判断一个文件是否可以被划分成多个数据块。

Hadoop提供了一些预定义的FileInputFormat类，如TextInputFormat用于读取文本文件，SequenceFileInputFormat用于读取SequenceFile格式的文件等，用户也可以通过继承FileInputFormat自定义输入格式类。

3.2.3 TextInputFormat

TextInputFormat是FileInputFormat默认的实现类，按行读取每条记录

key为该行的起始字节偏移量，为LongWritable类型

value 为这一行的内容，不包括终止符，为Text类型

3.2.4 CombineTextInputFormat

TextInputFormat是按文件进行规划分片，不管文件有多小，都是是一个单独的切片，这样会产生大量的MapTask，效率低下

CombineTextInputFormat用于小文件过多的场景，可以将多个小文件在逻辑上划分到一个切片

决定哪些块放入同一个分片时，CombineTextInputFormat会考虑到节点和机架的因素，所以在MR作业处理输入的速度不会下降

CombineTextInputFormat不仅可以很好的处理小文件，在处理大文件时也有好处，因为它在每个节点生成了一个分片，分片可能又多个块组成，CombineTextInputFormat使map操作中处理的数据量和HDFS中文件块的大小的耦合度降低了

3.3 MapReduce工作机制

3.3.1 MapTask工作机制

读取输入数据：MapTask通过InputFormat获得RecordReader，从输入InputSplit中解析出KV
Map阶段：将解析出的KV交给map()函数处理，产生一系列新的KV
Collect收集：数据处理完成之后，会调用OutputCollector.collect()输出结果。在该函数的内部，会生成KV分区，写入环形缓冲区中
Spill阶段：环形缓冲区满了之后，MR会将数据写到本地磁盘，形成一个临时文件，在写入之前，会对数据进行一次排序
merge阶段：所有数据处理完毕之后，MapTask会对所有临时文件进行一次合并，确保只生成一个数据文件

Spill阶段详情：

通过快速排序对环形缓冲区内的数据进行排序，先按照partition(后面会介绍)编号进行排序，然后再按照K进行排序。排序过后，数据以分区为单位聚集，分区内的所有数按照K有序
按照分区编号由小到大将分区数据写入工作目录下的临时文件 output/spillN.out(N表示当前溢写的次数)，如果设置了combiner(后面会介绍)，则写入文件之前，还会将分区中的数据进行一次聚集操作
将分区数据的元数据写入到内存索引数据结构SpillRecord中，每个分区的元数据包括临时文件的偏移量、压缩前后的数据大小，如果内存索引大于1MB，会将内存索引写到文件 output.spillN.out.index中

3.3.2 Partition分区

在MapReduce计算模型中，Map任务会将生成的键值对按照键进行排序，并将其划分到不同的分区中。分区的数量通常等于Reduce任务的数量。具体来说，Map任务会按照Partitioner函数定义的分区规则对键值对进行划分。Partitioner函数将每个键值对映射到一个分区编号，然后Map任务将其输出到对应的分区中。

Partitioner函数通常是由用户自定义实现的，其作用是将键值对映射到一个特定的分区。Hadoop提供了默认的Partitioner实现，即HashPartitioner，它将键哈希后取模得到分区编号，从而实现对键值对的划分。在实际应用中，用户可以根据自己的需求自定义Partitioner函数，以便将键值对划分到特定的分区中。

如果ReduceTask数量 > Partition数量，会产生多个空的输出文件

如果ReduceTask数量 < Partition数量，会导致有分区的数据无处安放，会Exception

如果ReduceTask数量 = 1，则不管有多少个分区文件，最终都会只产生一个文件

3.3.3 Combiner合并

Combiner在每个mapTask所在的节点运行

Combiner对每个MapTask的输出进行局部汇总，减少reduce阶段的负担

Combiner使用的前提是不能影响业务逻辑

3.3.4 ReduceTask工作机制

Copy：拉取数据，Reduce进程启动copy进程(Fetcher)，通过HTTP的方式请求maptask获取自己的文件，map task分区表示每个map task属于哪个reduce task

Merge：ReduceTask启动两个线程对内存和磁盘中的文件进行合并，防止文件过多。当内存中的数据达到一定阈值时，就会启动内存到磁盘的merge，与map端的相似。直到没有map端的数据才结束

合并排序：将数据合并成一个大数据，并进行排序

对排序后的数据调用reduce方法：对键相同的键值对调用reduce方法，每次调用会产生零个或多个键值对，最后将输出的键值对存入HDFS。

3.3.5 shuffle机制

Shuffle阶段的过程可以分为三个阶段：

Map端的输出：Map任务将生成的键值对按照键排序，并将其划分到不同的分区中。如果Map任务的输出缓存区已满，则需要将其溢出到本地磁盘的临时文件中。

数据传输：在Shuffle阶段中，Map任务的输出需要传输到Reduce任务所在的节点，以便Reduce任务可以从中提取和合并数据。数据传输是Shuffle阶段的关键步骤，其速度和效率直接影响整个MapReduce作业的性能。

Reduce端的输入：Reduce任务需要从本地磁盘读取属于自己的分区的临时文件，并对同一个分区中的键值对进行合并和排序。Reduce任务将合并后的结果输出到最终的输出文件中。

Shuffle阶段是MapReduce计算模型中非常重要的一个阶段，它的性能和效率对整个作业的执行时间和性能影响非常大。因此，优化Shuffle阶段的性能和效率是MapReduce应用程序优化的一个关键方向。

3.3.6 排序的解释

MapTask和ReduceTask对key进行排序是为了方便后续的数据处理和计算。

具体来说，对于MapTask而言，对输出的key进行排序可以将具有相同key值的记录聚合在一起，方便ReduceTask进行处理。

而对于ReduceTask而言，对输入的key进行排序可以让具有相同key值的记录相邻排列，方便进行聚合和计算。

一般来说，在Map任务中，对键值对进行快速排序的次数是一次，即将数据写入环形缓冲区之前对其中的键值对进行排序。这是因为，对于同一个Map任务的输出，在Map输出的环形缓冲区中进行快速排序即可满足Reduce任务在Shuffle阶段的需求，而不需要进行额外的排序。

在Shuffle阶段，如果存在多个环形缓冲区需要合并，Reduce任务会对它们进行归并排序。这是因为，不同Map任务的输出在Shuffle阶段需要合并，而这些输出之间的顺序是无序的，因此需要进行排序以便进行合并。这次排序是对整个数据集进行的，而不是对单个Map任务的输出进行的。

当Reduce任务接收到来自多个Map任务的中间结果时，它会对同一个分区内的所有键值对进行排序。这里采用的排序算法一般也是归并排序，因为归并排序的时间复杂度是O(nlogn)，且适合对大量数据进行排序。这次排序也是对整个数据集进行的，而不是对单个Map任务的输出进行的。

因此，总体来说，Shuffle阶段需要进行多次排序，具体排序的次数可能因具体实现而有所不同。但无论是哪种具体实现，Shuffle阶段都需要对整个数据集进行排序以便后续的计算和处理。

4 数据压缩

4.1 压缩算法对比

4.2 压缩位置选择

到此这篇关于Java大数据开发Hadoop MapReduce的文章就介绍到这了

java