Hadoop MapReduce

Posted alen-fly

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Hadoop MapReduce相关的知识,希望对你有一定的参考价值。

一、概述

1.是Hadoop中的一套分布式的计算框架

2.将整个计算过程拆分为2个阶段:Map阶段、Reduce阶段

3.Map阶段一般负责数据的整理,Reduce阶段负责数据的汇总

4.如果输入路径是一个文件,则MapReduce只处理这个文件;如果输入的是一个目录,则处理这个目录下的所有文件

 --注意:如果文件以 _ 开头,则该文件会被跳过,在MapReduce中,_ 开头的文件被认为是隐藏文件不需要处理

5.Rduce中的迭代器采用的是地址复用机制

6.Reduce中的迭代器只能遍历一次

7.在MapReduce中,针对Reduce出去的结果文件内容,如果不指定,内容(键值)中的默认键值之间用 进行分割的

9.在MapReduce中,如果需要实现的功能不需要Reduce,即Reduce没有业务逻辑,可以省略Reduce

二、序列化

1.在MapReduce中,要求被传输的数据必须能够被序列化

2.Hadoop中,序列化机制默认使用AVRO,但是Hadoop对AVRO的序列化机制进行了进一步的封装,提供了更简单的序列化机制

3.在Hadoop想要实现序列化,需要实现Writable,重新其中的方法

4.在Hadoop中序列化的时候,要求属性不能为null

序列化示例:

 

package com.apple.flow;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;

public class Flow implements Writable{
    
    private String phone;
    private String name;
    private String addr;
    //--在使用包装类,注意初始值的问题(null)
    private int flow;
    
    /*
     * 序列化方法
     * 如果是String,则调用writeUTF
     * 其他的:WriteInt,WriteLong,WriteByte,WriteDouble等
     */
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeUTF(phone);
        out.writeUTF(name);
        out.writeUTF(addr);
        out.writeInt(flow);
        
    }
    /*
     * 反序列化一定要注意顺序,和序列化的顺序一致
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        this.phone=in.readUTF();
        this.name=in.readUTF();
        this.addr=in.readUTF();
        this.flow=in.readInt();
        
    }
    public String getPhone() {
        return phone;
    }
    public void setPhone(String phone) {
        this.phone = phone;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getAddr() {
        return addr;
    }
    public void setAddr(String addr) {
        this.addr = addr;
    }
    
    public int getFlow() {
        return flow;
    }
    public void setFlow(int flow) {
        this.flow = flow;
    }
    @Override
    public String toString() {
        return "Flow [phone=" + phone + ", name=" + name + ", addr=" + addr + ", flow=" + flow + "]";
    }    

}

 

三、分区

1.分区在MapReduce中用于进行数据的分类

2.在MapReduce中,如果不指定,则默认只有1个分区

3.每一个分区都必须对应一个ReduceTask,每一ReduceTask都会产生一个结果文件

4.在MapReduce中对分区进行了编号,编码默认从0开始递增

5.分区的顶级父类是Partitioner

6.在MapReduce中,默认使用HashPartitioner

代码示例: 

public class FlowPartitioner extends Partitioner<Text,Flow>{

    @Override
    public int getPartition(Text key, Flow value, int numPartitions) {
        if(value.getAddr().equals("bj")){
            return 0;
        }
        else if(value.getAddr().equals("sh")){
            return 1;
        }else{
            return 2;
        }
                
    }

}

 

主类Driver入口

public class FlowDriver {

    public static void main(String[] args) throws Exception {
        Configuration conf=new Configuration();
        Job job=Job.getInstance(conf);
        
        job.setJarByClass(FlowDriver.class);
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Flow.class);
        
        job.setOutputKeyClass(Flow.class);
        job.setOutputValueClass(NullWritable.class);
        
        //--有3个分区,故有3个ReduceTask
        job.setNumReduceTasks(3);
        
        //--设置自定义的分区组件。如果不设定,默认用的是HashPartitioner
        //--默认的分区组件,会按Mapper输出key的hashcode分区,
        //--确保相同的key落到同一个分区里
        job.setPartitionerClass(FlowPartitioner.class);
        
        FileInputFormat.setInputPaths(job,
                new Path("hdfs://192.168.150.137:9000/flow"));
        
        FileOutputFormat.setOutputPath(job,
                new Path("hdfs://192.168.150.137:9000/flow/result"));
        
        job.waitForCompletion(true);
    }
}

Map类:

public class FlowMapper extends Mapper<LongWritable,Text,Text,Flow>{
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Flow>.Context context)
            throws IOException, InterruptedException {
        String line=value.toString();
        Flow f=new Flow();
        String[] info=line.split(" ");
        f.setPhone(info[0]);
        f.setName(info[1]);
        f.setAddr(info[2]);
        f.setFlow(Integer.parseInt(info[3]));
        
        context.write(new Text(f.getName()),f);
    }
}

Reduce类:

public class FlowReducer extends Reducer<Text,Flow,Flow,NullWritable>{

    @Override
    protected void reduce(Text key, Iterable<Flow> values, 
            Reducer<Text, Flow, Flow, NullWritable>.Context context)
            throws IOException, InterruptedException {
        Flow result=new Flow();
        for(Flow value:values){
            result.setPhone(value.getPhone());
            result.setName(value.getName());
            result.setAddr(value.getAddr());
            result.setFlow(result.getFlow()+value.getFlow());
            
        }
        
        context.write(result,NullWritable.get());
    }
}

 四、排序

1.在MapReduce中,会对键做自动的排序 - 自然排序

2.如果自定义一个类产生的对象想要作为键,那么这个对象必须要允许被排序 - 实现WritableComparable接口

3.多属性排序的场景称之为二次排序

以上是关于Hadoop MapReduce的主要内容,如果未能解决你的问题,请参考以下文章

Hadoop每日一讨论整理版

怎么在hadoop上部署mapreduce

Linux巩固记录 hadoop 2.7.4下自己编译代码并运行MapReduce程序

hadoop mapreduce的日志输出到哪里

如何在hadoop环境下执行mapreduce任务

Hadoop之Mapreduce详解