hadoop离线day09 mapreduce入门和wordcount示例

1391-王同学

发表文章数:51

首页 » 大数据 » 正文

1、mapreduce入门

分布式文件计算系统,主要用于计算我们的一些数据
MapReduce的核心思想:分而治之
最主要有两个阶段:
map阶段:负责任务拆分,
reduce阶段:负责结果聚合

mapreduce编程可控的八个步骤(天龙八部)

  • map阶段两个步骤
    1、第一步:读取文件,解析成key,value对,这里是我们的K1 V1
    2、第二步:接收我们的k1 v1,自定义我们的map逻辑,然后转换成新的key2 value2 进行输出 往下发送 这里发送出去的是我们k2 v2

  • shuffle阶段四个步骤
    3、第三步:分区 相同key的value发送到同一个reduce里面去,key合并,value形成一个集合
    4、第四步:排序 默认按照字段顺序进行排序
    5、第五步:规约
    6、第六步:分组

  • reduce阶段两个步骤
    7、接收我们的k2 v2 自定义我们的reduce逻辑,转换成新的k3 v3 进行输出
    8、将我们的K3 v3 进行输出

2、wordcount示例

需求:在一堆给定的文本文件中统计输出每一个单词出现的总次数
数据格式准备如下:

cd /export/servers
vim wordcount.txt
#hello,world,hadoop
#hive,sqoop,flume,hello
#kitty,tom,jerry,world
hadoop
hdfs dfs -mkdir /wordcount/
hdfs dfs -put wordcount.txt /wordcount/

java 实现代码

package cn.itcat.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * 除了javaweb的程序打成一个war包进行运行,其他的程序都是打成一个jar包进行运行
 * 运行一个jar包,需要一个main方法,作为我们程序的入口类
 */
public class MainCount extends Configured implements Tool{


    /**
     * 程序的入口类
     * @param args
     */
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        //这里执行完成之后,得到一个int类型的返回值,表示我们程序的退出状态码
        //如果退出状态码是0  程序执行成功
        //通过这里设置configuration,就相当于我们把父类的configuration设置值了
        int run = ToolRunner.run(configuration, new MainCount(), args);
        System.exit(run);


    }

    /**
     * 这个run方法很重,这里面就是通过job对象来组装我们的程序,说白了就是组装我们的八个类
     * @param args
     */
    @Override
    public int run(String[] args) throws Exception {
        //第一步:读取文件,解析成key,value对
        //从父类里面获取configuration配置文件
        //getInstance需要两个参数,第一个参数是我们的configuration配置文件,第二个参数叫做jobName随便写
        Job job = Job.getInstance(super.getConf(), "xxx");

        //如果打包到集群上面去运行,需要添加这一句,指定我们main方法所在的java类
        job.setJarByClass(MainCount.class);

        TextInputFormat.addInputPath(job,new Path("hdfs://node01:8020/wordcount"));
        job.setInputFormatClass(TextInputFormat.class);

        job.setMapperClass(WordCountMapper.class);

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


        job.setReducerClass(WordCountReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        TextOutputFormat.setOutputPath(job,new Path("hdfs://node01:8020/wordcountoutput"));

        job.setOutputFormatClass(TextOutputFormat.class);
        boolean b = job.waitForCompletion(true);
        return b?0:1;
    }
}
package cn.itcat.wordcount;

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;


public class WordCountMapper extends Mapper<LongWritable,Text,Text,IntWritable> {
    /***
     * 重写map方法,实现我们自己的逻辑,接受我们key1,value1  转换成新的k2  v2  输出
     * @param key  注意这个key是我们的k1
     * @param value  注意这个value是我们的v1
     * @param context  上下文对象,承上启下,衔接我们的上面的组件与下面的组件
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//hive,sqoop,flume,hello
        //第一步:切开我们一行数据
        String line = value.toString();
        String[] split = line.split(",");//[hive,sqoop,flume,hello]
        //  key2    value2    往下发送
        //  hive   1
        for (String word : split) {

           Text k2 =  new Text(word);
           IntWritable v2 =  new IntWritable(1);

           //通过write方法,将我们的数据往下发送
            context.write(k2,v2);

        }


    }
}
package cn.itcat.wordcount;

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

import java.io.IOException;

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

    /**
     *
     * @param key  注意这个key  是  k2
     * @param values  注意这个values是一个集合,集合的类习惯是  v2的类型
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int j= 0;
        for (IntWritable value : values) {
            //IntWritable这个类,没有普通的  +  方法,不能累加,需要转换成  int类型进行累加
            int num = value.get();
            j += num;

        }
        //输出我们的key3  value3  类型
        context.write(key,new IntWritable(j));


    }
}

未经允许不得转载:作者:1391-王同学, 转载或复制请以 超链接形式 并注明出处 拜师资源博客
原文地址:《hadoop离线day09 mapreduce入门和wordcount示例》 发布于2020-12-09

分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录