MapReduce天龙八部
2024-04-21 06:37:52
天龙八部：
map阶段：
第一步：读取文件，解析成key，value对，形成k1，v1；
第二步：接受k1，v1，自定义逻辑，转换成新的k2，v2，进行输出
shuffle阶段：
第三步：分区：
相同key的value值发送到同一个reduce中去，key合并，value形成一个集合
第四步：排序 默认按照自字段顺序进行排序
第五步：规约
第六步：分组
reduce阶段
第七步：接受k2，v2 自定义reduce逻辑，转换成新的k3，v3进行输出
第八步：将k3，v3进行输出
注意：
上面八个任务都是单独的一个java类，
八个步骤完了以后，通过job任务组装mr程序，
进行任务的提交
最简单的MapReduce程序：
主类：
package MR;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.Mapper;
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;public class MainCount extends Configured implements Tool {public static void main(String[] args) throws Exception {//这里执行完成后，得到一个int类型的返回值，表示我们的程序退出状态码//如果退出的是0，程序执行成功//这里设置configuration，相当于我们把父类的configuration设置值了int run = ToolRunner.run(new Configuration(), new MainCount(), args);System.exit(run);}/*必须实现run方法，这里面就是通过job对象来组装我们程序，组装八个类*/@Overridepublic int run(String[] strings) throws Exception {//第一步：读取文件，解析成key，value//获取一个job，job的作用主要是用来组装MapReduce各个阶段的任务//Job的getInstance方法需要两个参数，第一个是一个Configuration的实例，第二个是jobName，可以自己随便定义Job job = Job.getInstance(super.getConf(), "xxx");//把job和需要操作的文件所在的目录添加进来，注意此时增加的目录TextInputFormat.addInputPath(job,new Path("hdfs://hadoop11:8020/wordcount"));//打包到集群，必须添加如下配置，不添加的话会报错job.setJarByClass(MainCount.class);//指定job需要执行的任务的原始数据的类型job.setInputFormatClass(TextInputFormat.class);//第二步：自定义map类//指定mapjob.setMapperClass(WordCountMapper.class);//设置k2，v2的类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//第七步：自定义reduce类//指定reducejob.setReducerClass(WordCountReducer.class);//设置k3.v3job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//第八步：输出文件//指定路径TextOutputFormat.setOutputPath(job,new Path("hdfs://hadoop11:8020/wordcountoutput"));//设置job输出数据的类型job.setOutputFormatClass(TextOutputFormat.class);//提交任务到集群boolean b = job.waitForCompletion(true);//使用三元表达确认返回值return b?0:1;}
}
定义一个map类：
package MR;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;
//定义四个泛型
/*
k1--行偏移量
v1--行内容
k2--单词内容
v2--单词的次数*/
public class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {/***必须重写map方法，实现逻辑* @param key LongWritable key--代表k1* @param value Text value--代表v1* @param context Context context--上下文对象，将数据往下发送* @throws IOException* @throws InterruptedException*/@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//第一步：将行数据进行切分,用逗号进行分割String[] words = value.toString().split(",");//遍历切分的文字，通过write方法向下发送for(String word:words){context.write(new Text(word),new IntWritable(1));}}
}
定义一个reduce类
package MR;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;
//reduce的泛型参数代表k2，v2，k3，v3
public class WordCountReducer extends Reducer<Text, IntWritable,Text,IntWritable> {/***和map类同理，必须重写reduce方法* @param key Text key--k2* @param values Iterable<IntWritable> values--是一个集合，v2的类型* @param context 承上启下，向下发送* @throws IOException* @throws InterruptedException*/@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {//sum用来统计values的个数int sum=0;for (IntWritable value:values){// ntWritable是一个对象，而如果想要用int参加计算，那么我们需要调用IntWritable对象的一个成员get()，这个函数返回int.sum+=value.get();}context.write(key,new IntWritable(sum));}
}
MapReduce程序运行模式
本地运行模式
（1）mapreduce程序是被提交给LocalJobRunner在本地以单进程的形式运行
（2）而处理的数据及输出结果可以在本地文件系统，也可以在hdfs上
（3）怎样实现本地运行？写一个程序，不要带集群的配置文件
本质是程序的conf中是否有mapreduce.framework.name=local以及yarn.resourcemanager.hostname=local参数
（4）本地模式非常便于进行业务逻辑的debug，只要在eclipse中打断点即可
本地模式运行代码设置
configuration.set(“mapreduce.framework.name”,“local”);
configuration.set(“yarn.resourcemanager.hostname”,“local”);
TextInputFormat.addInputPath(job,new Path(“file:///F:\传智播客大数据离线阶段课程资料\3、大数据离线第三天\wordcount\input”));
TextOutputFormat.setOutputPath(job,new Path(“file:///F:\传智播客大数据离线阶段课程资料\3、大数据离线第三天\wordcount\output”));
集群运行
程序运行，将项目打包，发送至hadop集群上执行
hadoop jar original-day03_hdfs-1.0-SNAPSHOT.jar.0 MR.MainCount
MapReduce分区和排序
分区的思路：
第一步：读取文件，形成k1–Longwritab，v1–text
第二步：自定义map接收k1，v1 转换成k2–text，v2–LongWritable
第三步：自定义一个分区类：
使用getPartition（）方法
对k2进行分割，得到一个数组，使用数组下标得到一个result值
使用result值作为分区依据
>result 返回 分区号
<result 返回分区号
第四步：自定义reduce类接受k2，v2 直接写出去即可，转换成k3–text，v3–Nullwritab直接输出
简单的案例：
主类：
package mr.demo1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;public class PartitionMain extends Configured implements Tool {@Overridepublic int run(String[] args) throws Exception {Job job = Job.getInstance(super.getConf(), "mypartiton");job.setJarByClass(PartitionMain.class);TextInputFormat.addInputPath(job,new Path(args[0]));job.setInputFormatClass(TextInputFormat.class);job.setMapperClass(PartitionMapper.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(NullWritable.class);//第三步job.setPartitionerClass(PartitonOwn.class);job.setReducerClass(PartitionReduce.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);//设置redeuce的数量，不设置不行//reduce的个数和分区数相同job.setNumReduceTasks(2);job.setOutputFormatClass(TextOutputFormat.class);TextOutputFormat.setOutputPath(job,new Path(args[1]));boolean b = job.waitForCompletion(true);return b?0:1;}public static void main(String[] args) throws Exception {int run = ToolRunner.run(new Configuration(), new PartitionMain(), args);System.exit(run);}}
mapper类
package mr.demo1;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class PartitionMapper extends Mapper<LongWritable, Text,Text, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {context.write(value,NullWritable.get());}
}
reduce类：
package mr.demo1;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class PartitionReduce extends Reducer<Text, NullWritable,Text,NullWritable> {@Overrideprotected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {context.write(key,NullWritable.get());}
}
分区类：
package mr.demo1;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;public class PartitonOwn extends Partitioner<Text, NullWritable> {@Overridepublic int getPartition(Text text, NullWritable nullWritable, int numReduceNum) {String line = text.toString();String[] split = line.split("\t");if(Integer.parseInt(split[5])>15){return 0;}else {return 1;}}}
排序的思路：
第一步：读取文件，解析成k1–LongWritable，v1–text
第二步：自定义javaBean，实现WritableComparable，接受k1，v1
转换成k2–JavaBean，v2–Nullwritable
第三步：接受k2，v2不做处理，直接输出
主类：
package mrSort.demo1;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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;public class SortMain extends Configured implements Tool {@Overridepublic int run(String[] strings) throws Exception {Job job = Job.getInstance(super.getConf(), "sort");//解析文件job.setInputFormatClass(TextInputFormat.class);TextInputFormat.addInputPath(job,new Path("file:///E:\\baishi\\测试目录\\input\\sort.txt"));//自定义map逻辑job.setMapperClass(SortMapper.class);job.setMapOutputKeyClass(k2Bean.class);job.setMapOutputValueClass(NullWritable.class);//自定义reducejob.setReducerClass(SortReduce.class);job.setOutputKeyClass(k2Bean.class);job.setOutputValueClass(NullWritable.class);job.setOutputValueClass(TextOutputFormat.class);TextOutputFormat.setOutputPath(job,new Path("file:///E:\\baishi\\测试目录\\output04"));boolean b = job.waitForCompletion(true);return b?0:1;}public static void main(String[] args) throws Exception {Configuration configuration=new Configuration();System.exit(ToolRunner.run(configuration,new SortMain(),args));}}k2bean排序类：
package mrSort.demo1;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class k2Bean implements WritableComparable<k2Bean>{/*封装两个字段*/private String first;private Integer second;/*用于数据的比较排序*/@Overridepublic int compareTo(k2Bean o) {/*int a=compareto(b);如果a>b 返回1；如果a=b，返回0；如果a<b,返回-1；所以比较两个字段比较的话，只能用大于0，等于0，小于0表示*///当前的值和传进来的o进行比较//如果第一个字段相同，就比较第二个字段//如果第二个字段相同，就直接返回结果int i = this.first.compareTo(o.first);if(i ==0){//比较第二个字段int i1 = second.compareTo(o.second);return i1;}else {return i;}}/*序列化*/@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeUTF(first);dataOutput.writeInt(second);}/*反序列化*/@Overridepublic void readFields(DataInput dataInput) throws IOException {this.first=dataInput.readUTF();this.second=dataInput.readInt();}public String getFirst() {return first;}public void setFirst(String first) {this.first = first;}public Integer getSecond() {return second;}public void setSecond(Integer second) {this.second = second;}@Overridepublic String toString() {return first +"\t"+second;}
}
mapper类
package mrSort.demo1;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class SortMapper extends Mapper<LongWritable, Text,k2Bean, NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//自定义计数器，计数map阶段输入多少条数据Counter counter = context.getCounter("mr_intput_count", "map_total_result");counter.increment(1L);String[] split = value.toString().split("\t");k2Bean k2Bean = new k2Bean();k2Bean.setFirst(split[0]);k2Bean.setSecond(Integer.parseInt(split[1]));context.write(k2Bean,NullWritable.get());}
}
reduce类
package mrSort.demo1;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;public class SortReduce extends Reducer<k2Bean, NullWritable,k2Bean,NullWritable> {//定义一个类public static enum Counter{REDUCE_INPUT_RECORD,REDUCE_OUTPUT_RECORD,}@Overrideprotected void reduce(k2Bean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {//定义计数器，计数reduce输出条数org.apache.hadoop.mapreduce.Counter counter = context.getCounter(Counter.REDUCE_INPUT_RECORD);counter.increment(1L);for (NullWritable value:values) {org.apache.hadoop.mapreduce.Counter counter1 = context.getCounter(Counter.REDUCE_OUTPUT_RECORD);counter1.increment(1L);//将排序后的数据进行输出context.write(key, NullWritable.get());}}
}
规约的过程
发生在reduce阶段，接收k2，v2
combiner不改变数据结果，只用于调优，减少发送到reduce的数据量
思路：
第一步：读取文件，解析k1–LongWritable，v1-text
第二步：自定义javabean，转换成k2–javabean，v2–Nullwritab，输出
第三步：自定义一个Combiner规约类，接受k2，v2 ，直接输出
注释：因为combine类和reduce类一直，所以一般直接调用reduce类
job.setCombinerClass(SortReduce.class);
第四步：自定义reduce类直接输出
案例：MapReduce案例
MapReduce天龙八部相关推荐
- 分布式计算引擎MapReduce
应用场景 当数据量大到一定程度,传统的技术无法进行解决的时候,那么需要采用分布式计算引擎MapReduce来尝试解决了! 操作步骤 1. 架构图 Hadoop是由Apache基金会所开发的分布式系统基 ...
- 04 分布式文件系统以及MapReduce入门程序
文章目录 04 分布式文件系统以及MapReduce入门程序 分布式文件系统HDFS 1.分布式文件系统hdfs详细介绍 2.HDFS分布式文件系统设计目标 3.HDFS的来源 4.hdfs的架构图 ...
- 【hadoop 学习笔记-3】wordcount 统计小说天龙八部中汉字出现的频率
说明 天龙八部小说,3MB+ 中文 140W字以上. 用wordcount 统计一下每个不同的汉字 / 词汇出现的次数. Mapper package com.test;import org.ap ...
- mapreduce理解_大数据
map:对不同的数据进行同种操作 reduce:按keys 把数据规约到一起 看这篇文章请出去跑两圈,然后泡一壶茶,边喝茶,边看,看完你就对hadoop 与MapReduce的整体有所了解了. [前言 ...
- 2021年大数据Hadoop（二十二）：MapReduce的自定义分组
全网最详细的Hadoop文章系列,强烈建议收藏加关注! 后面更新文章都会列出历史文章目录,帮助大家回顾知识重点. 目录 本系列历史文章 前言 MapReduce的自定义分组 需求 分析 实现 第一步: ...
- 第2节 mapreduce深入学习：4, 5
第2节 mapreduce深入学习:4.mapreduce的序列化以及自定义排序 序列化(Serialization)是指把结构化对象转化为字节流. 反序列化(Deserialization)是序列化 ...
- 第一个MapReduce程序
计算文件中每个单词的频数 wordcount 程序调用 wordmap 和 wordreduce 程序. 1 import org.apache.hadoop.conf.Configuration; ...
- hadoop程序MapReduce之SingletonTableJoin
需求:单表关联问题.从文件中孩子和父母的关系挖掘出孙子和爷奶关系 样板:child-parent.txt xiaoming daxiong daxiong alice daxiong jack 输出: ...
- 初学Hadoop之图解MapReduce与WordCount示例分析
Hadoop的框架最核心的设计就是:HDFS和MapReduce.HDFS为海量的数据提供了存储,MapReduce则为海量的数据提供了计算. HDFS是Google File System(GFS) ...
最新文章
- 计算机复试题库.doc,计算机基础知识面试题库.doc
- openwrt安装oracle,Openwrt安装软件的方法-tomcat 随笔小记-install ubuntu 12.04 in virtualbox_169IT.COM...
- Matlab 2012安装图解
- 【软考-软件设计师】校验码
- java 安全库_国家信息安全漏洞库
- Android获取手机联系人或通讯录的基本信息（如姓名、电话）
- hive或mysql报错Too many connections
- mysql查询_MYSQL查询
- Eclipse的快捷键大全
- curl http_code状态码 含义
- Python按照索引访问list
- splay详解（二）
- 软件基本功：代码要有道理
- VB简易绘图板的实现
- 学生信息管理系统心得
- 如何完全禁止win10自动更新（自动升级）
- 解决MAC OS X不识别Kindle Fire
- 查看一个进程打开了哪些文件的命令
- Window10蓝牙无法连接的解决方案（已成功）
- 使用packet Tracer配置静态路由和动态路由