IT数码 购物 网址 头条 软件 日历 阅读 图书馆
TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		图片批量下载器
↓批量下载图片,美女图库↓ 		图片自动播放器
↓图片自动播放器↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁
 
   -> 大数据 -> 学习笔记Hadoop（五）—— MapReduce开发入门 -> 正文阅读

[大数据]学习笔记Hadoop（五）—— MapReduce开发入门

文章目录

一、MapReduce

https://img-blog.csdnimg.cn/823567368ae84eab919655d4d1661846.png
MapReduce是Google提出的一个软件架构，用于大规模数据集（大于1TB）的并行运算。概念“Map（映射）”和“Reduce（归纳）”，及他们的主要思想，都是从函数式编程语言借来的，还有从矢量编程语言借来的特性。

当前的软件实现是指定一个Map（映射）函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（归纳）函数，用来保证所有映射的键值对中的每一个共享相同的键组。

二、MapReduce开发环境搭建

环境准备： Java， Intellij IDEA， Maven
开发环境搭建方式

java安装链接及步骤：https://www.cnblogs.com/de-ming/p/13909440.html

2.1、Maven环境

在这里插入图片描述
添加依赖

https://search.maven.org/artifact/org.apache.hadoop/hadoop-client/3.1.4/jar

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述
添加源码
在这里插入图片描述
在这里插入图片描述

2.2、手动导入Jar包

Hadoop安装包链接：https://pan.baidu.com/s/1teHwnBH2Qm6F7iWZ3q-hSQ
提取码：cgnb

新建一个java工程
在这里插入图片描述
在这里插入图片描述
然后，搜JobClient.class，点击’Choose Sources’
在这里插入图片描述

这样就OK了，可以看到JobClient.java

三、MapReduce单词计数源码分析

3.1、打开WordCount.java

打开：https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-mapreduce-examples/3.1.4，复制Maven里面的内容
在这里插入图片描述
粘贴到源码
在这里插入图片描述
搜索WordCount
在这里插入图片描述

在这里插入图片描述

3.2、源码分析

3.2.1、MapReduce单词计数源码 : Map任务

在这里插入图片描述

3.2.2、MapReduce单词计数源码 : Reduce任务

在这里插入图片描述

3.2.3、MapReduce单词计数源码 : main 函数

设置必要参数及组装MapReduce程序在这里插入图片描述

四、MapReduce API介绍

  • 一般MapReduce都是由Mapper， Reducer 及main 函数组成。
  • Mapper程序一般完成键值对映射操作;
  • Reducer 程序一般完成键值对聚合操作;
  • Main函数则负责组装Mapper，Reducer及必要的配置;
  • 高阶编程还涉及到设置输入输出文件格式、设置Combiner、Partitioner优化程序等;

4.1、MapReduce程序模块 : Main 函数

在这里插入图片描述

4.2、MapReduce程序模块： Mapper

  • org.apache.hadoop.mapreduce.Mapper
    在这里插入图片描述

4.3、MapReduce程序模块： Reducer

  • org.apache.hadoop.mapreduce.Reducer
    在这里插入图片描述

五、MapReduce实例

5.1、流程（Mapper、Reducer、Main、打包运行）

  1. 参考WordCount程序，修改Mapper;
  2. 直接复制 Reducer程序；
  3. 直接复制Main函数，并做相应修改;
  4. 编译打包 ;
  5. 上传Jar包;
  6. 上传数据;
  7. 运行程序;
  8. 查看运行结果;

5.2、实例1：按日期访问统计次数:

1、参考WordCount程序，修改Mapper;
（这里新建一个java程序，然后把下面(1、2、3步代码)复制到类里）

    public static class SpiltMapper
            extends Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        //value: email_address | date
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            String[] data = value.toString().split("\\|",-1);  //
            word.set(data[1]);   //
            context.write(word, one);
        }
    }

2、直接复制 Reducer程序；

    public static class IntSumReducer
            extends Reducer<Text,IntWritable,Text,IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

3、直接复制Main函数，并做相应修改;

public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(CountByDate.class);   //我们的主类是CountByDate
        job.setMapperClass(SpiltMapper.class);  //mapper：我们修改为SpiltMapper
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job,
                new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

4、编译打包 (jar打包)

在这里插入图片描述
在这里插入图片描述

build出现错误及解决办法：
在这里插入图片描述
在这里插入图片描述

完成
在这里插入图片描述

5/6、上传jar包&数据
email_log_with_date.txt数据包链接：https://pan.baidu.com/s/1HfwHCfmvVdQpuL-MPtpAng
提取码：cgnb
在这里插入图片描述
上传数据包(注意开启hdfs)：
在这里插入图片描述
上传OK（浏览器：master:50070查看）
在这里插入图片描述

7、运行程序
(注意开启yarn)
在这里插入图片描述
上传完成后：

(master:8088)

在这里插入图片描述
8、查看结果
(master:50070)
在这里插入图片描述

5.3、实例2：按用户访问次数排序

Mapper、Reducer、Main程序
SortByCountFirst.Mapper

package demo;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class SortByCountFirst {
    //1、修改Mapper
    public static class SpiltMapper
            extends Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        //value: email_address | date
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            String[] data = value.toString().split("\\|",-1);
            word.set(data[0]);
            context.write(word, one);
        }
    }

    //2、直接复制 Reducer程序，不用修改
    public static class IntSumReducer
            extends Reducer<Text,IntWritable,Text,IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
        ) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    //3、直接复制Main函数，并做相应修改;
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: demo.SortByCountFirst <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "sort by count first ");
        job.setJarByClass(SortByCountFirst.class);   //我们的主类是CountByDate
        job.setMapperClass(SpiltMapper.class);  //mapper：我们修改为SpiltMapper
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job,
                new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

SortByCountSecond.Mapper

package demo;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

public class SortByCountSecond {
    //1、修改Mapper
    public static class SpiltMapper
            extends Mapper<Object, Text, IntWritable, Text> {

        private IntWritable count = new IntWritable(1);
        private Text word = new Text();
        //value: email_address \t count
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            String[] data = value.toString().split("\t",-1);
            word.set(data[0]);
            count.set(Integer.parseInt(data[1]));
            context.write(count,word);
        }
    }

    //2、直接复制 Reducer程序，不用修改
    public static class ReverseReducer
            extends Reducer<IntWritable,Text,Text,IntWritable> {

        public void reduce(IntWritable key, Iterable<Text> values,
                           Context context
        ) throws IOException, InterruptedException {
            for (Text val : values) {
                context.write(val,key);
            }
        }
    }

    //3、直接复制Main函数，并做相应修改;
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: demo.SortByCountFirst <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "sort by count first ");
        job.setJarByClass(SortByCountSecond.class);   //我们的主类是CountByDate
        job.setMapperClass(SpiltMapper.class);  //mapper：我们修改为SpiltMapper
//        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(ReverseReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job,
                new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

然后打包上传

yarn jar sortbycount.jar demo.SortByCountSecond -Dmapreduce.job.queuename=prod email_log_with_date.txt sortbycountfirst_output00

yarn jar sortbycount.jar demo.SortByCountSecond -Dmapreduce.job.queuename=prod email_log_with_date.txt sortbycountfirst_output00 sortbycountsecond_output00
  大数据 最新文章
实现Kafka至少消费一次
亚马逊云科技：还在苦于ETL？Zero ETL的时代
初探MapReduce
【SpringBoot框架篇】32.基于注解+redis实现
Elasticsearch：如何减少 Elasticsearch 集
Go redis操作
Redis面试题
专题五 Redis高并发场景
基于GBase8s和Calcite的多数据源查询
Redis——底层数据结构原理
上一篇文章      下一篇文章      查看所有文章
加:2021-09-29 10:21:27  更:2021-09-29 10:21:40 
 
开发: C++知识库 Java知识库 JavaScript Python PHP知识库 人工智能 区块链 大数据 移动开发 嵌入式 开发工具 数据结构与算法 开发测试 游戏开发 网络协议 系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑 笔记本 显卡 显示器 固态硬盘 硬盘 耳机 手机 iphone vivo oppo 小米 华为 单反 装机 图拉丁

360图书馆 购物 三丰科技 阅读网 日历 万年历 2025年2日历 -2025/2/23 13:17:14-

图片自动播放器
↓图片自动播放器↓ 		TxT小说阅读器
↓语音阅读,小说下载,古典文学↓ 		一键清除垃圾
↓轻轻一点,清除系统垃圾↓ 		图片批量下载器
↓批量下载图片,美女图库↓
  网站联系: qq:121756557 email:121756557@qq.com  IT数码