MapReduce之WordCount案例实操

 

目录

前期准备：

本机测试：

mapper阶段：

Reduce阶段：

Driver类：

集群测试：

---

前期准备：

因为MapReduce中案例比较多，所以需要单独创建一个工程

准备工作   创建工程后先改maven仓库的地址（创建工程后默认为idea自带的仓库**提示 在你打开别的项目后，在你重新打开本项目的时候，maven会改回idea的maven）

让后在   项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”

（打印INFO级别的日志）

填入：

log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

让后在Java包下创建三级目录  并且创建三个类（对应mapper，reduce,driver）

本机测试：

mapper阶段：

        mapper阶段继承自Mapper

  在类中重写map方法  在map方法外对     Text  和   IntWritable  进行实例化

代码：

package com.tangxiaocong.mapreduce.wordcount2;

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

import java.io.IOException;

/*
VALUEIN,        map阶段的value输入类型   text类型  一行
<KEYIN,         map阶段的key输入类型  LongWritable (偏移量)
KEYOUT,                输出的key  类型为text
VALUEOUT                输出的value  类型为int
>*/
public class WordCountMapper extends Mapper <LongWritable, Text,Text, IntWritable>{

    //定义属性 实例化 减少内存的消耗  在下边循环中的话  会循环创建 在全局可以多次使用
    private Text text = new Text();
    private IntWritable intWritable = new IntWritable(1);//map阶段不需要计算 同1为1
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //获取一行数据  转换成string
        String s = value.toString();
        //切割  切割后的单词存入数组
        String[] s1 = s.split(" ");
        //循环写出--输出  写出需要桥梁context这个抽象类
        for (String s2 : s1) {
            //数组中是String类型  需要转换成Text
            //封装text
            //此set是方法  不是Java中的接口  Java中有接口set 不可重复
            text.set(s2);
            //转换后通过context的write写出
            context.write(text,intWritable);


        }
    }
}

Reduce阶段：

        

package com.tangxiaocong.mapreduce.wordcount2;

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

import java.io.IOException;

public class WordCountReduce extends Reducer<Text, IntWritable,Text,IntWritable> {
       private IntWritable outv= new IntWritable();
    /*
    * Iterable<IntWritable> values   是一个集合的老祖宗   reduce阶段为会把同类集合化   两个<tangxiaoc,1>  reduce阶段会先合并成
    * tangxiaocong,(1,1)
    * */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            //Iterable<IntWritable> values  里现在是    (1,1)  现在需要把他们求和
        int sum =0;
        for (IntWritable value : values) {
            sum+=value.get();   //get方法是获取他的值
        }
        outv.set(sum);
        context.write(key,outv);
    }
}

Driver类：

package com.tangxiaocong.mapreduce.wordcount;

import org.apache.hadoop.conf.Configuration;
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;

import java.io.IOException;

//driver   其实是一个固定的套路
public class WordCountDriver {
    public static void main(String[] args) throws Exception {
        //1.获取job

        //Configuration  是job的配置信息类
        Configuration entries = new Configuration();
        Job job = Job.getInstance(entries);
        //2.设置jar包路径

        job.setJarByClass(WordCountDriver.class);  //一般通过全类名反射过去jar包的位置
        //3. 关联mapper和reducer

        job.setReducerClass(WordCountReduce.class);
        job.setMapperClass(WordCountMapper.class);
        //4.设置map输出的kv类型.

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        //5.设置最终的kv类型

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //6.设置输入路径和输出路径

        FileInputFormat.setInputPaths(job, new Path("D:\\hello.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\hellocount"));
        //7.提交job

        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);  //退出  成功返回0 失败返回1

    }
}

本次测试为本机测试，通过maven导入Hadoop的包来进行的   输入输出文件的路径为本地路径，而我们在企业开发中一般是在windows上编写，打包发送到Linux上，如果执行任务较多 后期会编写脚本执行程序。

集群测试：

下边重写driver类：只需要把输入输出的路径改成手动输入    输入输出路径

 

 进行打包

将不带依赖的包进行复制到Linux系统（Hadoop目录下）

使用hadoop jar  命令  执行jar包所在的本地系统中的项目  

（输出目录不能存在）

执行计算的过程中可以通过Hadoop yarn看到计算的资源调度的web页面