Hadoop--MapReduce运行处理流程

时间：2015-07-31 01:33:15 阅读：162 评论：0 收藏：0 [点我收藏+]

1. Hadoop 编程读写HDFS：
Hadoop文件API起点：FileSystem.
通过调用factory方法FileSystem.get(Configuration conf)来的到FileSystem实例.
具体写法：
Configuration conf = new Configuration();
FileSystem hdfs = FileSystem.get(conf);
得到一个专用于本地文件系统的FileSystem对象，可以用factory方法FileSystem.getLocal(Configuration conf);
FileSystem local = FileSystem.getLocal(conf);
Hadoop文件API使用Path编制文件和目录名. 使用FileStatus对象来存储元数据, PutMerge程序合并一个本地目录所有文件.
Path inputDir = new Path(args[0]);
FileStatus[] inputFiles = local.listStatus(inputDir);
PutMerge程序通过FSDataInputStream对象访问Path来读取文件.
FSDataInputStream in = local.open(inputFiles[i].getPath());
FSDataInputStream是一个Java标准输入流.

public class PutMerge{
public void test(String[] args){
Configuration conf = new Configuration();
FileSystem hdfs = FileSystem.get(conf);
FileSystem local = FileSystem.getLocal(conf);

//设定输入目录与输出文件
Path inputDir = new Path(args[0]);
Path hdfsFile = new Path(args[1]);

try{
//得到本地文件列表
FileStatus[] inputFiles = local.listStatus(inputDir);
//生成HDFS输出流
FSDataOutputStream out = hdfs.create(hdfsFile();

for(int i=0; i<inputFiles.length; i++){
System.out.println(inputFiles[i].getPath().getName());
//打开本地输入流
FSDataInputStream in = local.open(inputFiles[i].getPath());
byte buffer[] = new byte[256];
int bytesRead = 0;
while((bytesRead = in.read(buffer))>0)}
out.write(buffer, 0, bytesRead);
}
in.close();
}
}catch(Exception e){
e.printStackTrace();
}
}

}

程序答题流程： 1. 根据用户定义的参数设置本地目录和HDFS的目标文件.
2. 提取本地输入目录中每个文件的信息.
3. 创建一个输出流写入到HDFS文件.
4. 遍历本地目录中的每个文件，打开一个输入流来读取该文件.
5. 一个标准的Java文件复制过程.

2. MapReduce程序

MapReduce程序通过操作键值对来处理数据, 一般形式为：
map: (k1, v1) -> list(k2, v2)
reduce(k2, list(v2)) -> list(k3, v3)

技术分享

Hadoop数据类型：
MapReduce框架只支持序列化的类在框架中充当键或者值.
具体点，也就是说：实现Writable接口的类可以是值，而实现WritableComparable<T>接口的类既可以是键也可以是值.键在Reduce阶段会进行排序，值只会被简单地传递.

技术分享

实现WritableComparable接口的类

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.cmpareTo(o.arrivalNode);
}
}

3. Mapper
mapper需继承MapReduceBase基类并实现Mapper接口. mapper和reducer的基类均为MapReduceBase类.

MapReduceBase提供了
void configure(JobConf job)---该函数提取XML配置文件或主类中的参数，在数据处理前调用
void close() ---map任务结束前执行. 完成所有的结尾工作, 比如关闭数据库连接等

mapper类中的方法--map, 用于处理每一个map对应的数据分片, 也就是一个单独的键/值对.

技术分享

4. Reducer
reducer实现需继承MapReduceBase基类并实现Reduce接口，允许配置和清理.
reducer任务接受来自各个mapper的输出时，按照键/值对中的键对输入数据进行排序，并将相同见的值归并. 然后调用reduce()方法. 并通过迭代处理那些与指定键相关联的值，生成一个(可能为空)的列表(K3, V3). 最后由OutputCollector接收reduce阶段的输出，并写入输出文件.

技术分享

5. Partitioner

Partitioner：将mapper的结果输出给不同的reducer. 这就是partitioner的工作. Partitioner就是为了让输出的值mapper应用量身定制的Partitioner, 送往相同的reducer. 否则就会造成mapper输出的数据被处理两次，并且两次都是错误的.

实现Partitioner的类
public class Edge Implements WritableComparable<Edge>{

@Override
public int getPartition(Edge key, Writable value, int numPartitions) throws IOException{
return key.getDepartureNode().hashCode() % numPartitions;
}

@Override
public void configure(JobConf conf) {}

}

6. MapReduce处理的基本原则：
1. 将输入数据分割成块, 在Hadoop服务器集群中并行运行. 也就是被称为输入分片(Input Split)
2. 并行处理切分输入数据的原则. FSDataInputStream 支持随机读, 将每个文件块分为一个分片, 然后由不同驻留在每个分片的机器进行处理, 就可以自动实现并行.
3. MapReduce操作是基于键值/对的, 但是Hadoop也支持一些其他的数据各种, 并允许自定义格式.

总结： MapReduce框架，包含的操作： data splitting(数据分割), shuffling(洗牌), Partitioning(分组), Combining(合并), 当然还有最核心的Map 和Reduce 操作.

Hadoop--MapReduce运行处理流程

标签：hadoop mapreduce 工作流程原理

原文地址：http://blog.csdn.net/wild_elegance_k/article/details/47158851

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行