map reduce 实战

map reduce原理

问题：怎样解决海量数据的计算？

• MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题.
• MR由两个阶段组成：Map和Reduce，用户只需要实现map()和reduce()两个函数，即可实现分布式计算，非常简单。

执行步骤

map任务处理

• 读取输入文件内容，解析成key、value对。对输入文件的每一行，解析成key、value对。每一个键值对调用一次map函数。
• 写自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
• 对输出的key、value进行分区，
• 对不同分区的数据，按照key进行排序、分组。相同key的value放到一个集合中。
• (可选)分组后的数据进行归约。

reduce任务处理

• 对多个map任务的输出，按照不同的分区，通过网络copy到不同的reduce节点。
• 对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑，对输入的key、value处理，转换成新的key、value输出。
• 把reduce的输出保存到文件中

word count实例

image.png

map代码：

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

import java.io.IOException;

public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        //接收一行数据
        String line = value.toString();
        //分割
        String[] words = line.split(" ");
        //迭代
        for(String w : words){
            //发送
            context.write(new Text(w), new LongWritable(1));
        }
    }

    
}

reduce代码

import java.io.IOException;

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

public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

    @Override
    protected void reduce(Text key, Iterable<LongWritable> v2s, Context context)
            throws IOException, InterruptedException {
        //定义一行计数器
        long sum = 0;
        //迭代他的次数
        for(LongWritable lw : v2s){
            //求和
            sum += lw.get();
        }
        //输出
        context.write(key, new LongWritable(sum));
    }

    
}

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
/**
 * 1.定义一个类，这个类是MR执行入口
 * 2.定义一个类，这个类要继承org.apache.hadoop.mapreduce.Mapper，指定泛型的类型（k1,v1和k2,v2），然后重新map方法，接收数据，实现具体的业务逻辑
 * 3.定义一个类，这个类要继承org.apache.hadoop.mapreduce.Reducer，指定泛型的类型(k2,v2和k3,v3)，然后重新reduce方法，接收数据(k2和v2s，已经分组)，实现具体的业务逻辑
 * 4.在main方法中将自定义的mapper和reducer组装起来，提交
 * 5.打包
 * @author zx
 *
 */
public class WordCount {

    public static void main(String[] args) throws Exception {
        // 构造一个Job对象，job对象要依赖Configuration
        Configuration conf = new Configuration();
        conf.set("mapreduce.framework.name", "yarn");
        conf.set("yarn.resourcemanager.hostname", "itcast01");
        Job job = Job.getInstance(conf);
        
        //注意：要将main所在的类设置一下
        job.setJarByClass(WordCount.class);
        
        //设置Mapper相关的属性
        job.setMapperClass(WCMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        
        //设置Reducer相关属性
        job.setReducerClass(WCReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        job.setCombinerClass(WCReducer.class);
        
        //提交任务
        job.waitForCompletion(true);
        
        
        
    }
    
    

}

详细代码见：worcount

idea 打jar包

• 选中Java项目工程名称，在菜单中选择 File->project structure...

  
  image.png

• 在弹出的窗口中左侧选中"Artifacts"，点击"+"选择jar，然后选择"from modules with dependencies"。

image.png

• 在配置窗口中配置"Main Class"。

  
  image.png

• 配置“Directory for META-INF/MAINFEST.MF”， 放在resource目录下
• 配置完成后如下图所示，点击OK进入下一步。我个人建议选择“extract to the target JAR”，这样所有依赖的jar包都会放在生成的jar包中。

image.png

• 完成后，点击OK，Apply等按钮，回到IDEA的主菜单，选择“Build - Build Artifacts”下的“Build”或者“Rebuild”即可生成最终的可运行的jar，

上传wordCount.txt到hdfs： hadoop fs -put wordCount.txt

执行MR的命令：
hadoop jar <jar在linux的路径> <main方法所在的类的全类名> <参数>
例子：
hadoop jar mr.jar hdfs://server-2:9000/wordCount.txt /out2

得到结果：

 hadoop fs -cat /out2/part-r-00000
hard    1
hello   2
nihao   2
study   1
world   1
wusong  2
yes     1