MapReduce之WordCount

package com.uplooking.bigdata.mr.wc;

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 java.io.IOException;

/**

 * 统计hdfs上面的一个文件hello.txt中每一个单词出现的次数

 * hello.txt内容：

        hello wuhan

        hello xiao mei

        hello xiao hong

 map<LongWritable, Text, Text, IntWritable>

 reduce<Text, IntWritable, Text, IntWritable>

 */

public class WordCountApp {

    /*

    * 关于map和reduce的管理调度

    */

    public static void main(String[] args) throws Exception {//驱动代码 用于管理map和reduce

        //1、声明创建一个执行map和reduce的对象，这个对象我们一般称之为一个Job

        Configuration conf = new Configuration();

        String jobName = WordCountApp.class.getSimpleName();

        Job job = Job.getInstance(conf, jobName);

        //接下来的动作，将写好的代码打成一个jar

        //需要设置包含驱动程序的类的字节码

        job.setJarByClass(WordCountApp.class);

        //2°、设置map，使用FileInputFormat设置map的输入，需要知道hdfs文件的路径，同时需要告诉从属于哪一个job

        //2.1

        FileInputFormat.setInputPaths(job, "/hello.txt");

        //2.2

        job.setInputFormatClass(TextInputFormat.class);//使用相应的文件格式化方式进行解析

        //2.3设置用哪一个map类执行相应操作

        job.setMapperClass(WordCountMapper.class);

        //2.4设置map输出的key和value的类型

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(IntWritable.class);

        //3°、设置reduce

        //3.1 设置reduce从属于哪一个类

        job.setReducerClass(WordCountReducer.class);

        //3.2设置reduce的输出key和value的类型

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(IntWritable.class);

        //3.3 设置将reduce的结果输出到hdfs上的路径，以及采用何种格式化方式输出到hdfs

        job.setOutputFormatClass(TextOutputFormat.class);

        FileOutputFormat.setOutputPath(job, new Path("/out"));//如果输出路径已经存在，则会抛出目录存在异常

        //3.4设置有几个reduce来完成汇总

        job.setNumReduceTasks(1);

        //4°、提交程序

        job.waitForCompletion(true);

    }

    /**

     * map()相关

     * Mapper类的四个泛型参数

     * KEYIN---->k1--->当前行在文本中的偏移量,LongWritable

     * VALUEIN----->v1--->当前行的内容,Text

     * KEYOUT------>k2  Text

     * VALUEOUT------>v2 IntWritable

     */

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

       /*

       * 编写map阶段的业务逻辑

       * map()函数的前两个参数就是K1，v1

       *

       * */

        @Override

        protected void map(LongWritable k1, Text v1, Context context)

                throws IOException, InterruptedException {

            String line = v1.toString();//得到每一行的内容

            String[] splits = line.split(" ");//将字符串按照分隔符进行切分，得到字符串数组

            for(String word : splits) {//使用map阶段的上下文对象将这些数据写出到shuffle中

                Text k2 = new Text(word);

                IntWritable v2 = new IntWritable(1);

                context.write(k2, v2);//组装好数据写出

            }

        }

    }

    /**

     * reduce相关

     * <KEYIN---->k2    类型Text

     * VALUEIN ---->v2s  类型IntWritable

     * KEYOUT   ---->k3  类型Text

     * VALUEOUT ---->v3   类型IntWritable

     *

     */

    static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        @Override

        protected void reduce(Text k2, Iterable<IntWritable> v2s, Context context)

                throws IOException, InterruptedException {

            int sum = 0;

            for(IntWritable v2 : v2s) {//循环遍历v2s中的每一个数据

                sum += v2.get();

            }

            context.write(k2, new IntWritable(sum));//将汇总结果输出到hdfs文件中

        }

    }

}

将上述使用idea打包生成mr-wc.jar，上传到hadoop服务器/opt/jars/hadoop

使用参数化的命令

yarn jar /opt/jars/hadoop/mr-wc.jar com.uplooking.bigdata.mr.WordCountApp2 /hello /out