2021年大数据Hadoop（十七）：MapReduce编程规范及示例编写

<dependencies>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-common</artifactId>

            <version>2.7.5</version>

        </dependency>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-client</artifactId>

            <version>2.7.5</version>

        </dependency>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-hdfs</artifactId>

            <version>2.7.5</version>

        </dependency>

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-mapreduce-client-core</artifactId>

            <version>2.7.5</version>

        </dependency>

        <dependency>

            <groupId>junit</groupId>

            <artifactId>junit</artifactId>

            <version>4.12</version>

        </dependency>

        <dependency>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-simple</artifactId>

            <version>1.7.25</version>

        </dependency>

    </dependencies>

//首先要定义四个泛型的类型

//keyin:  LongWritable    valuein: Text

//keyout: Text            valueout:IntWritable

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

//map方法的生命周期：  框架每传一行数据就被调用一次

//key :  这一行的起始点在文件中的偏移量

//value: 这一行的内容

@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

    //拿到一行数据转换为string

    String line = value.toString();

    //将这一行切分出各个单词

    String[] words = line.split(" ");

    //遍历数组，输出<单词，1>

    for(String word:words){

        context.write(new Text(word), new LongWritable (1));

    }

}

}

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

//生命周期：框架每传递进来一个kv 组，reduce方法被调用一次

 @Override

 protected void reduce(Text key, Iterable<LongWritable > values, Context context) throws IOException, InterruptedException {

//定义一个计数器

int count = 0;

//遍历这一组kv的所有v，累加到count中

for(LongWritable value:values){

count += value.get();

}

context.write(key, new LongWritable (count));

 }

}

public class WordCountRunner {

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        //1、创建建一个job任务对象

        Configuration configuration = new Configuration();

        Job job = Job.getInstance(configuration, "wordcount");



        //2、指定job所在的jar包

        job.setJarByClass(WordCountRunner.class);



        //3、指定源文件的读取方式类和源文件的读取路径

        job.setInputFormatClass(TextInputFormat.class); //按照行读取

        //TextInputFormat.addInputPath(job, new Path("hdfs://node1:8020/input/wordcount")); //只需要指定源文件所在的目录即可

         TextInputFormat.addInputPath(job, new Path("file:///E:\\input\\wordcount")); //只需要指定源文件所在的目录即可



        //4、指定自定义的Mapper类和K2、V2类型

        job.setMapperClass(WordCountMapper.class); //指定Mapper类

        job.setMapOutputKeyClass(Text.class); //K2类型

        job.setMapOutputValueClass(LongWritable.class);//V2类型



        //5、指定自定义分区类（如果有的话）

        //6、指定自定义分组类（如果有的话）

        //7、指定自定义的Reducer类和K3、V3的数据类型

        job.setReducerClass(WordCountReducer.class); //指定Reducer类

        job.setOutputKeyClass(Text.class); //K3类型

        job.setOutputValueClass(LongWritable.class);  //V3类型



        //8、指定输出方式类和结果输出路径

        job.setOutputFormatClass(TextOutputFormat.class);

        //TextOutputFormat.setOutputPath(job, new  Path("hdfs://node1:8020/output/wordcount")); //目标目录不能存在，否则报错

        TextOutputFormat.setOutputPath(job, new  Path("file:///E:\\output\\wordcount")); //目标目录不能存在，否则报错



        //9、将job提交到yarn集群

        boolean bl = job.waitForCompletion(true); //true表示可以看到任务的执行进度



        //10.退出执行进程

        System.exit(bl?0:1);

    }

}