MapReduce学习+wordcount

1、关于mapreduce的定义：

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

2、mapreduce的优点（总的来说就是简单）

1. MapReduce易于编程
它简单的实现一些接口，就可以完成一个分布式程序，这个分布式程序可以分布到大量廉价的PC机器上运行。也就是说你写一个 分布式程序，跟写一个简单的串行程序是一模一 样的。就是因为这个特点使得MapReduce编程变得非常流行。
2.良好的扩展性
当你的计算资源不能得到满足的时候，你可以通过简单的增加机器来扩展它的计算能力。

3.高容错性

MapReduce设计的初衷就是使程序能够部署在廉价的PC机器上，这就要求它具有很高的容错性。比如其中-台机器挂了，它可以把上面的计算任务转移到另外一一个节点上运行,不至于这个任务运行失败,而且这个过程不需要人工参与，而完全是由Hadoop内部完成的。
4.适合PB级以上海量数据的离线处理
可以实现上千台服务器集群并发工作，提供数据处理能力。

3、mapreduce的缺点（总体来说就是慢）

1.不擅长实时计算
MapReduce无法像MySQL-样，在毫秒或者秒级内返回结果。
2.不擅长流式计算
流式计算的输入数据是动态的，而MapReduce的输入数据集是静态的， 不能动态变化。这是因为MapReduce 自身的设计特点决定了数据源必须是静态的。
3.不擅长DAG (有向图)计算
多个应用程序存在依赖关系，后一个应用程序的输入为前一个的输出。在这种情况下，MapReduce并不是不能做，而是使用后，每个MapReduce作业的输出结果都会写入到磁盘，会造成大量的磁盘IO，导致性能非常的低下。

4、mapreduce的核心思想

map是将所有的数据映射咱们想要的形式（很多个单词一的形式），之后reduce对数据进行处理（进行合并）

 

5、常用数据序列化类型：

 6、下面进行mapreduce的编写操作：

具体的代码如下（代码内有注释）：

WcMapper.java:

package wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

//Longwritable是行首在文件中的偏移量
public class WcMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
    private Text word=new Text();
    private IntWritable one=new IntWritable(1);
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
       //拿到这一行数据
        String line = value.toString();
        //按照空格进行切分
        String[] words=line.split(" ");
        //遍历数组，将单词变成（word，1）的形式
        for (String word :words)
        {
            this.word.set(word);
            context.write(this.word,this.one);
        }

    }
}

　WcReducer.java

package wordcount;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

//reduce输入的是map输出的
public class WcReducer extends Reducer<Text, IntWritable,Text, IntWritable> {
    private IntWritable total=new IntWritable();
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum=0;
        for(IntWritable value:values)
        {
            sum +=value.get();
        }
        //包装结果并输出
        total.set(sum);
        context.write(key,total);
    }
}

WcDriver.java

package wordcount;

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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Job;

import java.io.IOException;

public class WcDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1、获取一个job实例
        Job job=Job.getInstance(new Configuration());
        //2、设置我们的类路径
        job.setJarByClass(WcDriver.class);
        //3、进行map和reduce的设置
        job.setMapperClass(WcMapper.class);
        job.setReducerClass(WcReducer.class);
        //4、设置mapper和reducer的输出的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //5、设置输入和输出的数据
        FileInputFormat.setInputPaths(job,new Path(args[0]));
        FileOutputFormat.setOutputPath(job,new Path(args[1]));

        //6、提交我们的设置
        boolean b=job.waitForCompletion(true);
        System.exit(b ? 0:1);
    }

}

　之后找到其输入的文件，以及确定上输出的文件。

之后运行结果如下：

 文件运行成功：

原来的输入文件如下：