MapReduce程序基本架构

news2024/12/23 19:30:37

MapReduce程序是以（键/值）对的形式来处理数据的，即可以通过以下的形式来表示：

map: (K1,V1) ➞ list(K2,V2)

reduce: (K2,list(V2)) ➞ list(K3,V3)

不令人惊奇的是，这是一种超越一般数据的数据流表示形式。在本文中将讲述一个MapReduce程序每个阶段的更多的细节。图中表示的是一个表示完全的Hadoop流程的高级图表，我们将在一点点地介绍各个流程的功能。

值得注意的是：各个节点之间彼此交流的唯一时间是在“shuffle”阶段，这样的通讯限制很大程度上促进了程序的可拓展性。

在分析数据在某个阶段中的传递和处理过程之前，我们必须首先对Hadoop所支持的数据类型有所熟悉。

Hadoop数据类型

尽管在之前的讨论中我们没有涉及到（键/值）概念，但是MapReduce架构不允许程序中的一些参数为任意的类。打个比方来说，尽管我们的的确确在讨论一些键值时把他们当成integer，string以及其他等等的数据类型，但是他们实际上不是标准的Java对象（就像Integer，String等等）。这就是为什么MapReduce架构拥有自己定义的序列化键值对来处理集群中的计算这是因为这些类会更好地支持这种架构下的序列化。

特别注意的是，实现了Writable接口的类可以是值，而实现了WritableComparable<T>接口的类则既可以为键也可以为值。所以说WritableComparable<T>接口是Writable和java.lang.Comparable<T>的混合体。我们需要对键进行比较是因为它们在Reduce阶段会被排序，而值在这个时候只是简简单单地经过。

Hadoop已经有一系列已经定义好的实现了WritableComparable接口的类，这其中包括了封装了许多基本数据类型的包装类，就像下面所给的表中所显示的那样：

除此之外，你还可以通过实现WritableComparable或者Writable接口来创造自己的类，就像下面的例子所给的：

public class Edge implements WritableComparable<Edge>{

    private String departureNode;

    private String arrivalNode;

    public String getDepartureNode() { return departureNode;}

    @Override

    public void readFields(DataInput in) throws IOException {

        departureNode = in.readUTF();

        arrivalNode = in.readUTF();

    }

    @Override

    public void write(DataOutput out) throws IOException {

        out.writeUTF(departureNode);

        out.writeUTF(arrivalNode);

    }

    @Override

    public int compareTo(Edge o) {

        return (departureNode.compareTo(o.departureNode) != 0)? departureNode.compareTo(o.departureNode) : arrivalNode.compareTo(o.arrivalNode);

    }

}

这个类中实现了Writable中的readFields（）和write（）两个方法。之后的程序中使用Java中的DataInput和DataOutput类来实现类内容的序列化。该类还实现了Comparable接口中的compareTo（）方法，其中返回值为-1，0和1分别代表了Edge对象是比所给的Edge对象小，相等或者大的关系。

Mapper

为了实现mapper的功能，一个类可以继承于MapReduceBase类并且实现Mapper接口。MapReduce类是作为mapper和reducer的基础类，该类中包括了两种重要的方法，这两个方法在类中分别等效为构造函数和析构函数的功能：

①void configure(JobConf job)——在这个函数中，你可以通过XML文件提取出相关参数，或者在程序中的开始，即数据开始处理之前调用这个函数提取出相关参数。

②void close()——作为在map程序结束之前的最后一个步骤，这个函数必须做好所有的结尾工作——数据库的连接，打开文件等等。

Mapper接口负责数据处理，其利用Mapper<K1,V1,K2,V2>的形式（其中键值类分别是实现了WritableComparable和Writable）。其唯一的方法就是处理一对键值。

void map(K1 key,V1 value,OutputCollector<K2,V2> output,Reporter reporter) throws IOException

这个函数产生根据给定的(K1,V1)键值输入对一系列的(K2,V2)键值对。OutputCollector接受map处理后的结果，Reporter提供了记录mapper处理的额外信息。

Hadoop提供了一些实用的mapper类拓展，你可以在下面的表中看到：

Reducer

就像前面的mapper一样，reducer必须首先继承MapReduce类和实现Reducer接口。而在Reducer接口中有下面的一个方法：

void reduce(K2 key,Iterator<V2> values,OutputCollector<K3,V3> output,Reporter reporter

) throws IOException

当reducer任务收到不同mapper的输出时，它会对到来的数据根据键来排列然后汇总具有相同键的所有值。在这之后reduce（）函数就被调用，然后通过对值和对应的键进行迭代产生<K3,V3>对，OutputCollector接受reduce处理后的结果并把结果输出到一个文件中。Reporter同样提供记录reducer任务中的额外信息。

下面图表提供了一对Hadoop提供的reducer的实用类