正文 

当前位置： 首页 > 程序 > java教程

学习笔记hadoop——mapreduce开发入门(代码片段)

别呀 别呀     2023-01-07     606

关键词：

文章目录

一、MapReduce


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 hadoop学习笔记—4.初识mapreduce

一、神马是高大上的MapReduce　　MapReduce是Google的一项重要技术，它首先是一个编程模型，用以进行大数据量的计算。对于大数据量的计算，通常采用的处理手法就是并行计算。但对许多开发者来说，自己完完全全实现一个并行计...  查看详情  

                

hadoop学习笔记：java开发mapreduce


1. MapReduce的流程图（摘自马士兵老师视频），我们开发的就是其中的这两个（红框）过程。简述一下这个图，input就是我们需要处理的文件（datanode上文件的一个分块）；Split就是将这个文件进行拆分，默认的就是按照行来拆...  查看详情  

                

mapreduce学习笔记，理解学习hadoop的mapreduce计算系统


MapReduce概述：      MapReduce最早是在Google的论文中提出的，但是对应的代码并没有开源。从2004年Google公开发布MapReduce论文到2012为止，MapReduce已经成长为被广泛采用的分布式数据处理的业界标准。      MapReduce是...  查看详情  

                

hadoop学习笔记：使用mrjob框架编写mapreduce


1.mrjob介绍一个通过mapreduce编程接口（streamming）扩展出来的Python编程框架。2.安装方法pip install mrjob，略。初学，叙述的可能不是很细致，可以加我扣扣：2690382987，一起学习和交流~3.代码运行方式下面简介mrjob提供的3种代...  查看详情  

                

学习笔记hadoop——hadoop基础操作——mapreduce常用shell操作mapreduce任务管理(代码片段)


四、MapReduce常用Shell操作4.1、MapReduce常用ShellMapReduceShell此处指的是可以使用类似shell的命令来直接和MapReduce任务进行交互（这里不严格区分MapReduceshell及Yarnshell）。提交任务命令：yarnjar<jar>[mainClass]args...查看及修...  查看详情  

                

学习笔记hadoop——mapreduce编程进阶(代码片段)


...使用Partitioner优化程序3.1、自定义单词计数四、本地提交MapReduce程序4.1、自定义单词计数一、输出文件格式及序列化文  查看详情  

                

学习笔记hadoop（十五）——mapreduce编程进阶(代码片段)


...使用Partitioner优化程序3.1、自定义单词计数四、本地提交MapReduce程序4.1、自定义单词计数一、输出文件格式及序列化文  查看详情  

                

大数据零基础学习hadoop入门教程


...，具有可靠、高效、可伸缩的特点Hadoop的核心是YARN,HDFS,Mapreduce,常用模块架构如下?2、HDFS源自谷歌的GFS论文，发表于2013年10月，HDFS是GFS的克隆版，HDFS是Hadoop体系  查看详情  

                

hadoop入门容易吗？


...oop编程调用HDFS用Maven构建Mahout项目Mahout推荐算法API详解用MapReduce实现矩阵乘法从源代码剖析Mahout推荐引擎Mahout分步式程序开发基于物品的协同过滤ItemCFMahout分步式程序开发聚类KmeansPageRank算法并行实现三、案例分析海量Web日志分...  查看详情  

                

hadoop基础学习


...9;HDFS读数据完整流程（下载文件）HadoopMaReduce介绍MapReduce设计思想MapReduce介绍MapReduce特点MapReduce的局限性MapReduce实例进程MapReduce执行流程HadoopYARN介绍YARN的功能YARN架构、组件YARN交互流程YARN资源调度器schedule声明：本文...  查看详情  

                

大数据离线计算路线图-hadoop工程师,数据分析师


...ase框架的全面深入讲解，为了能轻松掌握相关知识，学习MapReduce开发的20个经典案例讲解以及部分Hadoop源代码的分析，借此深入学习内核原理。方法/步骤Zookeeper入门到精通视频教程详细讲解Zookeeper的安装配置、命令使用、存储结...  查看详情  

                

mapreducemapreduce基础入门


一、mapreduce入门 1、什么是mapreduce   首先让我们来重温一下hadoop的四大组件：HDFS：分布式存储系统MapReduce：分布式计算系统YARN：hadoop的资源调度系统Common：以上三大组件的底层支撑组件，主要提供基础工具包和RPC...  查看详情  

                

hadoop入门概念


...统基础架构。　　Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，则MapReduce为海量的数据  查看详情  

                

学习笔记hadoop——hadoop介绍


...oop任务调度和资源管理框架-YARN2.5、Hadoop分布式编程模型-MapReduce三、Hadoop生态环境3.1、ApacheHBase3  查看详情  

                

hadoop学习


...高速运算和存储。Hadoop的框架最核心的设计就是：HDFS和MapReduce。HDFS为海量的数据提供了存储，则MapReduce为海量的数据提供了计算。 Hadoop核  查看详情  

                

hadoop学习笔记-3


从jar包中提取默认配置core-default.xmlhadoop-common-<ver>.jarhdfs-default.xmlhadoop-hdfs-<ver>.jarmapred-default.xmlhadoop-mapreduce-client-core-<ver>.jaryarn-default.xmlhadoop-yarn-common-<  查看详情  

                

尚硅谷大数据hadoop教程-笔记01入门(代码片段)


...Hadoop教程-笔记02【HDFS】尚硅谷大数据Hadoop教程-笔记03【MapReduce】尚硅谷大数据Hadoop教程-笔记04【Yarn】尚硅谷大数据Hadoop教程-笔记04【生产调优手册】尚硅谷大数据Hadoop教程-笔记04【源码解析】目录00_尚硅谷大数据Hadoop课程整体...  查看详情  

                

spark学习入门


...、集成Hadoop、极高的活跃度。   Spark的速度比MapReduce快：MR计算模型太死板，而且里面最好性能的就是shuffle，shuffle中间的过程都是基  查看详情