MR案例之WordCount

1.在Eclipse上编写程序

1.1.导入hadoop的jar包

1.2.项目结构

WordCountMapper.java

package com.matrix.mr;

import java.io.IOException;

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

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    // 循环调用的方法，从split中每一行调用一次
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] words = value.toString().split(" ");
        for (int i = 0; i < words.length; i++) {
            String word = words[i];
            context.write(new Text(word), new IntWritable(1));
        }
    }

}

WordCountReduce.java

package com.matrix.mr;

import java.io.IOException;

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

public class WordCountReduce extends Reducer<Text, IntWritable, Text, IntWritable> {

    // reduce也是循环调用，每一组调用一次
    @Override
    protected void reduce(Text arg0, Iterable<IntWritable> arg1, Context arg2)
            throws IOException, InterruptedException {
        // 根据键进行分组，
        int sum = 0;

        for (IntWritable i : arg1) {
            sum = sum + i.get();
        }
        arg2.write(arg0, new IntWritable(sum));
    }

}

RunJob.java

	package com.matrix.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

public class RunJob {

    public static void main(String[] args) {
        // 读取配置文件
        Configuration conf = new Configuration();
        try {

            FileSystem fs = FileSystem.get(conf);

            Job job = Job.getInstance(conf);
            // 设置用户名称
            job.setJobName("wc");

            // 设置任务的入口类
            job.setJarByClass(RunJob.class);
            job.setMapperClass(WordCountMapper.class);
            job.setReducerClass(WordCountReduce.class);

            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);

            // 设置输入数据目录
            FileInputFormat.addInputPath(job, new Path("/usr/matrix/input/wc"));

            // 针对目录进行判断
            Path outdir = new Path("/usr/matrix/output/wc");
            if (fs.exists(outdir)) {
                fs.delete(outdir, true);
            }

            // 设置输出数据目录
            // path 一个目录 而且不能存在
            FileOutputFormat.setOutputPath(job, outdir);
            boolean f = job.waitForCompletion(true);

            if (f) {
                System.out.println("WordCount程序运行成功！");
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

wc.java

hadoop hello world
hello hive
hadoop world
hello hbase

2.导出jar包

3.将jar包上传至node1主机/opt/modules/hadoop-2.5.1目录下

3.1.在hdfs上创建目录

[root@node1 hadoop-2.5.1]# ./bin/hdfs dfs -mkdir -p /usr/matrix/input/wc

4.用命令运行jar包

[root@node1 hadoop-2.5.1]# hadoop jar wc.jar com.matrix.mr.RunJob

5.查看运行结果：

[root@node1 hadoop-2.5.1]# ./bin/hdfs dfs -cat /usr/matrix/output/wc/part-r-00000