运行一个mapreduce实例

本文改编自打开
因为参考文中步骤有部分运行不正确，所以自己记录下自己的步骤，并将原因整理了下。

Score.java文件

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
public class Score {
    public static class Map extends
            Mapper<LongWritable, Text, Text, IntWritable> {
        // 实现map函数
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            // 将输入的纯文本文件的数据转化成String
            String line = value.toString();
            // 将输入的数据首先按行进行分割
            StringTokenizer tokenizerArticle = new StringTokenizer(line, "\n");
            // 分别对每一行进行处理
            while (tokenizerArticle.hasMoreElements()) {
                // 每行按空格划分
                StringTokenizer tokenizerLine = new StringTokenizer(tokenizerArticle.nextToken());
                String strName = tokenizerLine.nextToken();// 学生姓名部分
                String strScore = tokenizerLine.nextToken();// 成绩部分
                Text name = new Text(strName);
                int scoreInt = Integer.parseInt(strScore);
                // 输出姓名和成绩
                context.write(name, new IntWritable(scoreInt));
            }
        }
    }



    public static class Reduce extends
            Reducer<Text, IntWritable, Text, IntWritable> {
        // 实现reduce函数
        public void reduce(Text key, Iterable<IntWritable> values,
                Context context) throws IOException, InterruptedException {
            int sum = 0;
            int count = 0;
            Iterator<IntWritable> iterator = values.iterator();
            while (iterator.hasNext()) {
                sum += iterator.next().get();// 计算总分
                count++;// 统计总的科目数
            }
            int average = (int) sum / count;// 计算平均成绩
            context.write(key, new IntWritable(average));
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // "localhost:9000" 需要根据实际情况设置一下
        conf.set("mapred.job.tracker", "localhost:9000");
        // 一个hdfs文件系统中的 输入目录 及 输出目录
        String[] ioArgs = new String[] { "input/score", "output" };
        String[] otherArgs = new GenericOptionsParser(conf, ioArgs).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: Score Average <in> <out>");
            System.exit(2);
        }

        Job job = new Job(conf, "Score Average");
        job.setJarByClass(Score.class);
        // 设置Map、Combine和Reduce处理类
        job.setMapperClass(Map.class);
        job.setCombinerClass(Reduce.class);
        job.setReducerClass(Reduce.class);
        // 设置输出类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 将输入的数据集分割成小数据块splites，提供一个RecordReder的实现
        job.setInputFormatClass(TextInputFormat.class);
        // 提供一个RecordWriter的实现，负责数据输出
        job.setOutputFormatClass(TextOutputFormat.class);
        // 设置输入和输出目录
        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

编译Score.java

javac Score.java

如果出现错误


vim /etc/profile
----------------

添加如下内容

#set hadoop environment
export HADOOP_HOME=/usr/local/hadoop
export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin
export CLASSPATH=$HADOOP_HOME/share/hadoop/common/hadoop-common-2.6.2.jar:$HADOOP_HOME/share/hadoop/mapreduce/hadoop-mapreduce-client-core-2.6.2.jar:$HADOOP_HOME/share/hadoop/common/lib/commons-cli-1.2.jar

注意：这里的环境变量中的classpath就是和原文的不同之处，也应该就是这个原因造成编译不通过

重新运行

javac Score.java

会生成三个class文件

ls | grep class

打成jar包

jar -cvf Score.jar ./Score*.class

这里也和原文不同，不是用tar命令打包，而应该是jar命令

新建路径/input/score

hadoop fs -mkdir  -p /user/root/input/scores

上传到Hadoop的HDFS

hadoop fs -put ./*.txt /input/score

查看上传结果

hadoop fs -ls -R /input/score

运行

hadoop jar Score.jar Score /input/score /output

格式
hadoop jar jar包所在目录类名称 HDFS中需要处理数据路径 HDFS中存放数据路径

这里如果报错，可能出现的情况有
http://blog.csdn.net/yeweiouyang/article/details/24467773

1、路径不对，注意当前路径和jar包所在的路径不一致
2、jar包有问题（一开始用的tar打包，所以出错尴尬）
3、jar包文件名错了

输出结果

hdfs dfs -ls output 或者 hadoop fs –ls –R /

查看结果

hdfs dfs -cat output/part-r-00000  或者 hadoop fs  -cat output/part-r-00000

参考文献
http://blog.sina.com.cn/s/blog_68cceb610101r6tg.html
也可以将上述操作的内容换成该文章中的WordCount.java源码以及file0和file1

秒客网