Hadoop MapReduce 作业思路详解

解析

第一步：寻求已知源码

首先我们拥有一个Wordcount代码：（无用引入已经抹除）

public class WordCount {
    public static class TokenizerMapper extends
            Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {

            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

    public static class IntSumReducer extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args)
                .getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: wordcount <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

我们可以看到这个源码的结构是这样的：

Map + Reduce + job(设置Mapper和Reducer)

于是我们的目标就是：分析Map和Reduce怎么魔改一下让他能用：

分析Mapper

public static class TokenizerMapper extends
        Mapper<Object, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();

    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {

        StringTokenizer itr = new StringTokenizer(value.toString());
        while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
        }
    }
}

首先我们看到：StringTokenizer（一个不认识的玩意）

网上搜索一下：

Java StringTokenizer 属于 java.util 包，用于分隔字符串。（菜鸟教程）

之后我们注意到一句代码：

context.write(word, one);

这显然是一句类似于写入的代码，写入到hadoop估计是给下一步处理的。

之后我们看看这个write传入了什么值：

一个是word,一个是one,我们把定义拿过来：

 private final static IntWritable one = new IntWritable(1);

private Text word = new Text();
word.set(itr.nextToken());

于是我们观察到这个东西的传值是：一个Text（word）和一个IntWritable（Int开头应该是说明数据流是Int的？）

并且我们根据wordcount的能力进行思考：

假设输入：

word word

wordcount输出：word 2

根据one的定义是new IntWritable(1),可以猜测它在Map处的处理应该是进行了两次context，每一次设置的都是1，也就是两次处理是加在一起的（注意这里还不能完全确定，但是后面Reduce处就能确定了）

之后我们便确定逻辑：

它首先分开字符串->之后每一个分开的字符串都会当一个单词被检测->检测到单词就context.write计数一个1->处理完毕。

之后我们看看Reducer是怎么回事。

分析Reducer

    public void reduce(Text key, Iterable<IntWritable> values,
            Context context) throws IOException, InterruptedException {
        int sum = 0;
        // Find values in IntWritable
        for (IntWritable val : values) {
        	//Add the sum.
            sum += val.get();
        }
        //Set the result for sum
        result.set(sum);
        //Now we have the truly result.
        context.write(key, result);
    }
}

（注释自己瞎加的不要在意）

我们首先看这个循环：

for (IntWritable val : values) {
        	//Add the sum.
            sum += val.get();
        }

也就是说不知道怎么来了一个IntWritable,然后获取里面的值加到sum里。

根据Map部分，这个IntWritable肯定是Map那里的那个context.write部分，也就是那个写死的new IntWritable(1)。

也就是说这就是统计单词数的。

同时我们能发现：这个values（循环的总部分）应该代表了相同单词对应的所有IntWritable集合。实现机制不知道，但是肯定是这样子，不然统计的sum不就不是这个单词的了吗？

再看剩余的代码就是设置结果和写入结果了。也就是说逻辑就是这样的：

Map（一次处理一行，处理完了之后把单词和一个数值1交上去）

Reducer（收集Map们，并且把相同单词的那个数值1加到一起，得到单词的总数）

于是我们便可以准备好对我们的数据进行处理了。

第二步：分析下一步的源码

首先我们传入的是CSV文件，我们用记事本打开看一下：

于是我们清楚了：

它是一行一行的；用逗号分开。

接下来我们便可以编写Map了：

想办法“沾点亲戚”？

首先我们看题：

注意到：

星期对应的架次是数字->星期相当于单词->只要出现了这个星期肯定有架次->一个变相的WordCount->写出一个输入自己数据的Wordcount.

注意到2：

航班号是用两个数据组合起来的数据->使用字符串分割可以得到这两个数据然后拼起来->飞行公里数对应一个后面的数值->如果得到航班号，每出现一次航班号就要加一次飞行公里数->假设飞行公里数是1，那么相当于航班号对应的加1，也就是说当飞行公里数为1时，等同于统计航班号的wordcount->注意到之前写死的new IntWritable(1)->将1改成飞行公里数->一个修改了更多的Wordcount。

现在我们已经搞懂了这玩意就是个变种的Wordcount，接下来我们就想办法和Wordcount"沾点亲戚"，让他达到要求。

开始魔改代码

首先我们可以看到上面的两个思路都不需要对reduce进行修改。也就是说都是Map的事情。

我们首先处理第一个：一个输入自己数据的Wordcount.

我们知道Java StringTokenizer 属于 java.util 包，用于分隔字符串。但是我们不会用啊！

这好办，反正我还知道有个分割方法我会：xxxx.toString().split(“xxx”)

于是我们改写代码：（请看代码的注释）

public static class TokenizerMapper extends
        Mapper<Object, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text weektext = new Text();
    private Text hangbantag = new Text();
    public void map(Object key, Text value, Context context)
            throws IOException, InterruptedException {
        //使用这种方法分割字符串，我们就能很方便的得到星期。
    	String[] csvvalue = value.toString().split(",");
    	String week = csvvalue[3];
    	weektext.set(week);
        //因为我们的目标还是出现了星期就一定有架次，所以我们只需要把上面的word换成我们得到的星期就行了
        //去掉表头
        if(!week.equals("DayOfWeek"))
    	context.write(weektext,one);
    }
}

于是这个代码就改好了，把输入文件丢进wcin之后试试看——

处理的数据已经一致了！说明思路无误！（最后那个是表头，想个办法去掉就行了，比如判断是不是和表头相同，如果是就不写入，这个好办）

接下来我们看另外一个：

复制一份刚刚的Map改个名字，然后魔改一番：（同样是看代码注释）

public static class DistanceMapper extends
    Mapper<Object, Text, Text, IntWritable> {
    	//FIX IT FOR MORE
    	//private final static IntWritable one = new IntWritable(1);
    	private Text hangbantag = new Text();
    	public void map(Object key, Text value, Context context)
        throws IOException, InterruptedException {
    		String[] csvvalue = value.toString().split(",");
    		String hangbandaihao = csvvalue[8];
    		String hangbanhao = csvvalue[9];
    		//组合在一起获取到了我们的航班号   
    		hangbantag.set(hangbandaihao + hangbanhao);
    		//得到飞行距离（字符串）
    		String distance = csvvalue[18];
            //使用TRY CATCH去掉表头（因为字符串转成数字会崩）
    		try
    		{
    			int distancenum = Integer.parseInt(distance);
                //将new IntWritable里的赋值直接填写成飞行距离，这样就完成了我们的思路
    			context.write(hangbantag,new IntWritable(distancenum));
    		}
    		catch(Exception e)
    		{
                //无关紧要……
    			System.out.println("Biao Tou");
    		}
}	

之后我们发现一个问题：两个Mapper，下面就一个Job……

所以我们要加一个Job。这个代码基本就是复刻一下，改一下输出路径（输出路径冲突不是报错嘛），然后改一下输入的Mapper:

//这边是原来wordcount的代码
      Job job = new Job(conf, "word count");
      job.setJarByClass(WordCount.class);
      //Set Mapper
      job.setMapperClass(TokenizerMapper.class);
      //Set combiner
      job.setCombinerClass(IntSumReducer.class);
      //Set Reducer
      job.setReducerClass(IntSumReducer.class);
      //Mapper output
      job.setOutputKeyClass(Text.class);
      //Reducer output
      job.setOutputValueClass(IntWritable.class);
      //Add files in this job（这里是添加输入和输出的位置）
      FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
      FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
      //这里是复制了上面之后开始修改的
      Job job2 = new Job(conf,"distance");//这里改了一下任务名
      job2.setJarByClass(WordCount.class);
//这里换了Mapper
      job2.setMapperClass(DistanceMapper.class);
      //Set combiner
      job2.setCombinerClass(IntSumReducer.class);
      //Set Reducer
      job2.setReducerClass(IntSumReducer.class);
      //Mapper output
      job2.setOutputKeyClass(Text.class);
      //Reducer output
      job2.setOutputValueClass(IntWritable.class);
      //这里修改了一下输出路径（那个"1"
      FileInputFormat.addInputPath(job2, new Path(otherArgs[0]));
      FileOutputFormat.setOutputPath(job2, new Path(otherArgs[1] + "1"));
//分析了一下，这是一个三目运算符
//原本是job.waitForCompletion(true)之后判断
//System.exit(job.waitForCompletion(true) ? 0 : 1);
//根据英语猜测应该是任务是否完成
      System.exit((job.waitForCompletion(true)&&job2.waitForCompletion(true)) ? 0 : 1);

运行结果：

其他的要求呢？

输出到文件已经有了代码，接下来就是想办法缝合到一起。

一些坑

你肯定认为“啊缝合一个老师给了的代码太简单了”，很不幸，代码是有问题的。

Configuration cfg = new Configuration();
FileSystem hdfs = FileSystem.get(cfg);
Path src = new Path(outputPath + "/part-00000");
Path dst = new Path("/root/flight.dat");
hdfs.copyToLocalFile(src, dst);

问题出在两行上：

1. FileSystem的获取问题：使用 FileSystem fs= ileSystem.get(conf); 获取fs会出现如下错误Error: java.lang.IllegalArgumentException: Wrong FS: hdfs:// …（原因我查询到的博客没有讲，烦请大佬指出）大多数网上说的做法是把集群上的core-site.xml和hdfs-site.xml放到当前工程下，但这不是正规处理。正确做法是使用srcPath生成FileSystem。
2. 第三行对应的文件名有错，输出的文件名为part-r-00000.

经过修改后的代码如下：

Configuration cfg = new Configuration(); 
//将src提前到FileSystem获取处，从而让下文可以得到src。
//修改文件名
Path src = new Path(outputPath + "/part-r-00000");
//使用srcPath生成FileSystem
FileSystem hdfs = src.getFileSystem(cfg);
Path dst = new Path("/root/flight.dat");
hdfs.copyToLocalFile(src, dst);

之后我们准备输出：

由于我们有两个要输出的表格，所以我们要复刻两份。之后我们修改最后System.exit()判断函数，最后的最后是这样的：

if(job.waitForCompletion(true)&&job2.waitForCompletion(true))
{
	String outputPath = otherArgs[1];
	Configuration cfg = new Configuration();
	Path src = new Path(outputPath + "/part-r-00000");
	FileSystem hdfs = src.getFileSystem(cfg);
	Path dst = new Path("/root/flight.dat");
	hdfs.copyToLocalFile(src, dst);
	//Distance output
	src = new Path(outputPath + "1" + "/part-r-00000");
	dst = new Path("/root/flight2.dat");
	hdfs = src.getFileSystem(cfg);
	hdfs.copyToLocalFile(src, dst);
}

总体代码如下：

package org.apache.hadoop.examples;

import java.io.IOException;
import java.util.StringTokenizer;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.FileSystem;

public class WordCount {
	//for mapper
    public static class TokenizerMapper extends
            Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text weektext = new Text();
        private Text hangbantag = new Text();
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
        	String[] csvvalue = value.toString().split(",");
        	// map the week and the hangbandaihao(fuck for no chinese commit)
        	String week = csvvalue[3];
        	String hangbandaihao = csvvalue[8];
        	weektext.set(week);
        	//hangbantag.set(hangbandaihao);
        	//System.out.println(week + " " + hangbandaihao);
            if(!week.equals("DayOfWeek"))
        		context.write(weektext,one);
        }
    }
    //For distance:
    public static class DistanceMapper extends
    Mapper<Object, Text, Text, IntWritable> {
    	//FIX IT FOR MORE
    	//private final static IntWritable one = new IntWritable(1);
    	private Text hangbantag = new Text();
    	public void map(Object key, Text value, Context context)
        throws IOException, InterruptedException {
    		String[] csvvalue = value.toString().split(",");
    		// Get the hangbandaihao + hangbanhao
    		String hangbandaihao = csvvalue[8];
    		String hangbanhao = csvvalue[9];
    		//Combine it then we can get the key.
    		hangbantag.set(hangbandaihao + hangbanhao);
    		//Get the distance.
    		String distance = csvvalue[18];
    		//18
    		try
    		{
    			int distancenum = Integer.parseInt(distance);
    			context.write(hangbantag,new IntWritable(distancenum));
    		}
    		catch(Exception e)
    		{

    		}
}	
}
    
   //For reduce
    public static class IntSumReducer extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            // Find values in IntWritable
            for (IntWritable val : values) {
            	//Add the sum.
                sum += val.get();
            }
            //Set the result for sum
            result.set(sum);
            //Now we have the truly result.
            context.write(key, result);
        }
    }
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args)
                .getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: wordcount <in> <out>");
           System.exit(2);
        }
        //A new Job
        Job job = new Job(conf, "word count");
        job.setJarByClass(WordCount.class);
        //Set Mapper
        job.setMapperClass(TokenizerMapper.class);
        //Set combiner
        job.setCombinerClass(IntSumReducer.class);
        //Set Reducer
        job.setReducerClass(IntSumReducer.class);
        //Mapper output
        job.setOutputKeyClass(Text.class);
        //Reducer output
        job.setOutputValueClass(IntWritable.class);
        //Add files in this job
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        //Another new job.
        Job job2 = new Job(conf,"distance");
        job2.setJarByClass(WordCount.class);
        job2.setMapperClass(DistanceMapper.class);
        //Set combiner
        job2.setCombinerClass(IntSumReducer.class);
        //Set Reducer
        job2.setReducerClass(IntSumReducer.class);
        //Mapper output
        job2.setOutputKeyClass(Text.class);
        //Reducer output
        job2.setOutputValueClass(IntWritable.class);
        //Reducer new output
        FileInputFormat.addInputPath(job2, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job2, new Path(otherArgs[1] + "1"));
        //Try to download the file on our disk
        if(job.waitForCompletion(true)&&job2.waitForCompletion(true))
        {
        	String outputPath = otherArgs[1];
        	//Download the file
        	Configuration cfg = new Configuration();
        	//He give us the wrong code which has IllegalArgumentException.
        	//And here he give us a wrong file name. 
        	Path src = new Path(outputPath + "/part-r-00000");
        	FileSystem hdfs = src.getFileSystem(cfg);
        	Path dst = new Path("/root/flight.dat");
        	hdfs.copyToLocalFile(src, dst);
        	//Distance output
        	src = new Path(outputPath + "1" + "/part-r-00000");
        	dst = new Path("/root/flight2.dat");
        	hdfs = src.getFileSystem(cfg);
        	hdfs.copyToLocalFile(src, dst);
        }
    }
}

（以上代码已去除表头）