MapReduce
1、优缺点
- 优点:易于编程、良好的扩展性、高容错性
- 缺点:不适合做实时计算、不适合做流式计算(要求数据是静态的)、不适合DAG(有向图)计算
复制代码 2、MapReduce的阶段分类【掌握】
MapReduce的步调在运行的过程中,一样平常分为两个阶段:Map阶段和Reduce阶段
2.1、第一阶段:Map
第一阶段,也称之为Map阶段。这个阶段会有若干个MapTask实例,完全并行运行,互不干系。每个MapTask会读取分析一个InputSplit(输入分片,简称分片)对应的原始数据。计算的结果数据会暂时生存到地点节点的本地磁盘里。
该阶段的编程模型中会有一个map函数需要开辟人员重写,map函数的输入是一个<key,value>对,map函数的输出也是一个<key,value>对,key和value的范例需要开辟人员指定。
2.2、第二阶段:Reduce
第二阶段,也称为Reduce阶段。这个阶段会有若干个ReduceTask实例并发运行,互不干系。但是他们的数据依靠于上一个阶段的所有MapTask并发实例的输出。一个ReudceTasK会从多个MapTask运行节点上fetch自己要处理的分区数据。颠末处理后,输出到HDFS上。
该阶段的编程模型中有一个reduce函数需要开辟人员重写,reduce函数的输入也是一个<key,value>对,reduce函数的输出也是一个<key,value>对。这里要夸大的是,reduce的输入其实就是map的输出,只不过map的输出颠末shuffle技术后变成了<key,List>而已。
注意:MapReduce编程模型只能包罗一个map阶段和一个reduce阶段,假如用户的业务逻辑非常复杂,那就只能多个MapReduce步调,串行执行。
3、IDE 编写MapReduce 实现wordCount
3.1、在IDE中需要依次创建WordCountMapper、WordCountReducer、WordCountDriver
示例中的包路径:com.ms.mshadoop.mapreduce
- package com.ms.mshadoop.mapreduce;
- 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;
- /**
- * Map阶段开始的时候,每一个MapTask都会逐行读取分片中的数据,并将读取到的数据进行扭转形成 keyIn, valueIn
- * 由于计算程序需要到不同节点之间进行移动,因此涉及到 keyIn/valueIn/keyOut/valueOut都必须支持序列化的类型
- * Hadoop提供了一套序列化的机制 writeable
- * byte => ByteWritable
- * short => ShortWritable
- * int => IntWritable
- * long => LongWritable
- * float => FloatWritable
- * double => DoubleWritable
- * boolean => BooleanWritable
- * String => Text
- * <p>
- * keyIn:读取到的 行数据 中 首字母 的偏移量 需要设计为LongWritable
- * valueIn:读取到的 行数据 需要设计为Text
- * <p>
- * keyOut:经过逻辑处理后,需要写出的键值对中 键的类型 Text
- * valueOut:经过逻辑处理后,需要写出的键值对中 值的类型 IntWritable
- */
- public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
- /**
- * 每当读取到一行数据时,将其扭转为keyIn和valueIn,调用该方法
- * @param key 行偏移量
- * @param value 行记录
- * @param context 操作上下文
- * @throws IOException
- * @throws InterruptedException
- */
- @Override
- protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
- //1、将读取的行数据,切割出每一个单词
- // 正则表达式中\s匹配任何空白字符,包括空格、制表符、换页符等等, 等价于[\f\n\r\t\v]
- // 而\s+则表示匹配任意多个上面的字符
- String[] words = value.toString().split("\\s+");
- //2、遍历每一个单词
- for (String word : words) {
- //3、为每一单词配上(value,1) 组成键值对 写出
- context.write(new Text(word),new IntWritable(1));
- }
- }
- }
复制代码- package com.ms.mshadoop.mapreduce;
- import org.apache.hadoop.io.IntWritable;
- import org.apache.hadoop.io.Text;
- import org.apache.hadoop.mapreduce.Reducer;
- import java.io.IOException;
- /**
- * keyIn: Map阶段输出的键类型
- * valueIn: Map阶段输出的值类型
- * <p>
- * keyOut: 最终输出的键值对中,键的类型
- * valueOut: 最终输出的键值对中,值的类型
- */
- public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
- /**
- * Reduce阶段逻辑处理
- *
- * @param key 键,Map阶段输出的键
- * @param values 输入进这个方法之前,MapReduce会按照键进行分组,将相同的键对应的所有值聚合到一起
- * @param context 上下文
- * @throws IOException
- * @throws InterruptedException
- */
- @Override
- protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
- //1、定义一个变量,来记录单子出现的总次数
- int timeCount = 0;
- for (IntWritable value : values) {
- timeCount += value.get();
- }
- context.write(key, new IntWritable(timeCount));
- }
- }
复制代码- package com.ms.mshadoop.mapreduce;
- 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;
- /**
- * 驱动类:主要功能
- * 1、创建Job
- * 2、Job属性配置
- * 3、Job提交
- */
- public class WordCountDriver {
- public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
- //1、创建Job
- Configuration conf = new Configuration();
- conf.set("fs.defaultFS", "hdfs://hadoop01:9820");
- Job job = Job.getInstance();
- //2、设置Job属性
- //在MapReduce程序中,用于处理Map任务的类
- job.setMapperClass(WordCountMapper.class);
- //在MapReduce程序中,用于处理Reduce任务的类
- job.setReducerClass(WordCountReducer.class);
- //在MapReduce程序中,用于处理驱动的类
- job.setJarByClass(WordCountDriver.class);
- //Map阶段输出的键值对的类型
- job.setMapOutputKeyClass(Text.class);
- job.setMapOutputValueClass(IntWritable.class);
- //Reduce阶段输出的键值对的类型
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(IntWritable.class);
- //设置输入和输出的路径
- FileInputFormat.setInputPaths(job, new Path(args[0]));
- FileOutputFormat.setOutputPath(job, new Path(args[1]));
- //3、提交任务
- System.exit(job.waitForCompletion(true) ? 0 : -1);
- }
- }
复制代码 注意:需要天生多个文件数量时,代码中增长
- //设置ReduceTask的数量,决定了最终生成的文件数量。这个数量最好和分区的数量保持一致。
- /**
- * 1、如果ReduceTask的数量多于分区的数量:会出现多余的ReduceTask空占资源,
- * 不去处理任何的数据,浪费资源,且生成空的结果文件
- * 2、如果ReduceTask的数量少于分区的数量:会出现某个分区的数据暂时无法处理,
- * 需要等待某ReduceTask任务处理结束后再1处理这个分区的数据,无法高效并发
- */
- job.setNumReduceTasks(2);
复制代码 3.2 执行步调
- 1、IDE中将项目打包成jar,上传到linux 的hadoop集群中
- 2、运行程序
- # hadoop jar 打包上传到服务器的jar 程序中WordCountDriver的包路径 /输入文件路径 /输出文件路径
- hadoop jar MapReduceApi-1.0-SNAPSHOT.jar com.ms.mshadoop.mapreduce.WordCountDriver /input /output
复制代码 4、分区器(Partitioner)
需求:将单词a-g开头的存一个文件,h-o存一个文件,其他存一个
4.1、在IDE中需要依次创建WordCountMapper、WordCountReducer、WordCountDriver、WordCountPartitioner,其中WordCountMapper、WordCountReducer与3.1相同
- package com.ms.mshadoop.partitioner;
- import org.apache.hadoop.io.IntWritable;
- import org.apache.hadoop.io.Text;
- import org.apache.hadoop.mapreduce.Partitioner;
- /**
- * Partitioner : 分区器,可以将map阶段输出的键值对,按照规则进行分区
- * 泛型,对应的是map方法输出的键值对类型
- */
- public class WordCountPartitioner extends Partitioner<Text, IntWritable> {
- /**
- * 每一个map方法输出的键值对,都会调用该方法进行分区,来确定分区号
- * @param text map 阶段输出的 键
- * @param intWritable map 阶段输出的 值
- * @param i ReduceTask 的数量
- * @return 分区号,从零开始,且必须连续
- */
- @Override
- public int getPartition(Text text, IntWritable intWritable, int i) {
- char c = text.toString().charAt(0);
- if (c >= 'a' && c <= 'g') {
- return 0;
- } else if (c >= 'h' && c <= 'o') {
- return 1;
- } else {
- return 2;
- }
- }
- }
复制代码 WordCountDriver中增长如下代码
- //应用分区器
- job.setPartitionerClass(WordCountPartitioner.class);
- //设置ReduceTask的数量,决定了最终生成的文件数量。这个数量最好和分区的数量保持一致。
- /**
- * 1、如果ReduceTask的数量多于分区的数量:会出现多余的ReduceTask空占资源,
- * 不去处理任何的数据,浪费资源,且生成空的结果文件
- * 2、如果ReduceTask的数量少于分区的数量:会出现某个分区的数据暂时无法处理,
- * 需要等待某ReduceTask任务处理结束后再1处理这个分区的数据,无法高效并发
- */
- job.setNumReduceTasks(3);
复制代码 5、IDE运行MapReduce的模式
- 在IDEA中运行MapReduce的程序,可以选择计算资源,也可以选择文件系统
- 计算资源:mapreduce.framework.name
- local:使用本地计算资源(CPU、内存)
- yarn:使用集群的计算资源
- 文件系统:fs.defaultFS
- hdfs://hadoop01:9820 使用分布式文件系统
- file:/// 使用本地文件系统
- 本地调试时可使用local模式进行,正常推荐使用yarn模式进行
复制代码 5.1、local 模式测本地文件
原理:
1、将MapReduce的任务资源调度设置为local,倒霉用YARN举行资源调度
2、将文件系统设置为本地文件系统,倒霉用HDFS
- Configuration conf = new Configuration();
- conf.set("mapreduce.framework.name","local");//设置为本地运行模式,任务不会在YARN上运行
- conf.set("fs.defaultFS","file:///");//设置为本地文件系统,不使用HDFS。
- Job job = Job.getInstance(conf );
- //设置输入和输出的路径
- FileInputFormat.setInputPaths(job, new Path("/输入:本地文件路径"));
- FileOutputFormat.setOutputPath(job, new Path("/输出:本地文件路径"));
复制代码 5.2、local 模式测集群文件
原理:
1、将MapReduce的任务资源调度设置为local,倒霉用YARN举行资源调度。
2、将文件系统设置为HDFS
- System.setProperty("HADOOP_USER_NAME", "root"); //设置hadoop的操作用户
- Configuration conf = new Configuration();
- conf.set("mapreduce.framework.name","local");//设置为本地运行模式,任务不会在YARN上运行
- conf.set("fs.defaultFS","hdfs://hadoop01:9820");//设置为分布式文件系统
- Job job = Job.getInstance(conf );
- //设置输入和输出的路径
- FileInputFormat.setInputPaths(job, new Path("/输入:hdfs文件路径"));
- FileOutputFormat.setOutputPath(job, new Path("/输出:dhfs文件路径"));
复制代码 5.3、YARN模式测集群
原理:
1、将MapReduce的任务资源调度设置为YARN
2、将文件系统设置为HDFS
- System.setProperty("HADOOP_USER_NAME", "root"); //设置hadoop的操作用户
- //1、创建Job
- Configuration conf = new Configuration();
- //local 模式测本地文件
- /*conf.set("mapreduce.framework.name","local");//设置为本地运行模式,任务不会在YARN上运行
- conf.set("fs.defaultFS","file:///");//设置为本地文件系统,不使用HDFS。*/
- //local 模式测集群文件
- /*conf.set("mapreduce.framework.name","local");//设置为本地运行模式,任务不会在YARN上运行
- conf.set("fs.defaultFS","hdfs://hadoop01:9820");//设置为分布式文件系统*/
- //yarn 模式测集群文件
- conf.set("mapreduce.framework.name", "yarn");//设置为本地运行模式,任务不会在YARN上运行
- conf.set("yarn.resourcemanager.hostname", "hadoop01");//设置ResourceManager
- conf.set("fs.defaultFS", "hdfs://hadoop01:9820");//设置为分布式文件系统
- conf.set("yarn.app.mapreduce.am.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的ApplicationMaster(AM)启动时的环境变量。
- conf.set("mapreduce.map.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的Map启动时的环境变量。
- conf.set("mapreduce.reduce.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的Reduce启动时的环境变量。
- conf.set("hadoop.security.authentication","simple");
- conf.set("mapreduce.app-submission.cross-platform","true");//跨平台提交任务
复制代码 注意事项:
在上述的配置都完成后,将步调打jar包,然后将jar包添加到classpath,才可以运行步调
- 在jar包上右键-> Add As Library
复制代码 6 Hadoop 序列化
- package com.ms.mshadoop.writable;
- import org.apache.hadoop.io.Text;
- import org.apache.hadoop.io.Writable;
- import java.io.DataInput;
- import java.io.DataOutput;
- import java.io.IOException;
- import java.io.Serializable;
- public class Student implements Writable, Serializable {
- private String name;
- private int age;
- // private Text remarks;
- public Student() {
- }
- public Student(String name, int age, Text remarks) {
- this.name = name;
- this.age = age;
- // this.remarks = remarks;
- }
- /**
- * 序列化:将内存对象,序列化为一个字节序列
- * @param dataOutput
- */
- @Override
- public void write(DataOutput dataOutput) throws IOException {
- //该方法中依次将所有属性序列化
- //当属性非hadoop中的序列类型时
- //将String 序列化
- dataOutput.writeUTF(name);
- //将int 序列化
- dataOutput.writeInt(age);
- //当属性为hadoop中的序列化类型时
- // remarks.write(dataOutput);
- }
- /**
- * 反序列化:将字节序列转 内存对象
- * 注意:反序列化时,读取属性顺序 必须 和序列化时顺序一致
- * @param dataInput
- */
- @Override
- public void readFields(DataInput dataInput) throws IOException {
- //当属性非hadoop中的序列类型时
- //将String 反序列化
- name = dataInput.readUTF();
- //将int 反序列化
- age = dataInput.readInt();
- //当属性为hadoop中的序列化类型时
- // remarks.readFields(dataInput);
- }
- }
- package com.ms.mshadoop.writable;
- import org.apache.hadoop.io.Text;
- import java.io.*;
- public class Test {
- public static void main(String[] args) throws IOException {
- Student student = new Student("张三丰",100,new Text("11111111"));
- File file = new File("C:\\Users\\xazyh\\Desktop\\1111.txt");
- ObjectOutputStream outputStream = new ObjectOutputStream(new FileOutputStream(file));
- outputStream.writeObject(student);
- outputStream.close();
- File file1= new File("C:\\Users\\xazyh\\Desktop\\2222.txt");
- DataOutputStream dataOutputStream = new DataOutputStream(new FileOutputStream(file1));
- student.write(dataOutputStream);
- dataOutputStream.close();
- }
- }
复制代码
7 Mapreduce 实现统计流量
- package com.ms.mshadoop.phoneFlow;
- import org.apache.hadoop.io.Writable;
- import java.io.DataInput;
- import java.io.DataOutput;
- import java.io.IOException;
- public class PhoneFlowBean implements Writable {
- private String phone;
- private int flowDown;
- private int flowUp;
- public PhoneFlowBean() {
- }
- public PhoneFlowBean(String phone, int flowDown, int flowUp) {
- this.phone = phone;
- this.flowDown = flowDown;
- this.flowUp = flowUp;
- }
- @Override
- public void write(DataOutput dataOutput) throws IOException {
- dataOutput.writeUTF(phone);
- dataOutput.writeInt(flowDown);
- dataOutput.writeInt(flowUp);
- }
- @Override
- public void readFields(DataInput dataInput) throws IOException {
- phone = dataInput.readUTF();
- flowDown = dataInput.readInt();
- flowUp = dataInput.readInt();
- }
- public int getSumFlow() {
- return flowDown + flowUp;
- }
- public String getPhone() {
- return phone;
- }
- public void setPhone(String phone) {
- this.phone = phone;
- }
- public int getFlowDown() {
- return flowDown;
- }
- public void setFlowDown(int flowDown) {
- this.flowDown = flowDown;
- }
- public int getFlowUp() {
- return flowUp;
- }
- public void setFlowUp(int flowUp) {
- this.flowUp = flowUp;
- }
- }
复制代码- package com.ms.mshadoop.phoneFlow;
- import org.apache.hadoop.io.LongWritable;
- import org.apache.hadoop.io.Text;
- import org.apache.hadoop.mapreduce.Mapper;
- import java.io.IOException;
- public class PhoneFlowMapper extends Mapper<LongWritable, Text, Text, PhoneFlowBean> {
- @Override
- protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
- String line = value.toString();
- //根据数据格式进行数据分割
- String[] split = line.split("\t");
- //手机号
- String phone = split[0];
- //下行流量
- String flowDown = split[1];
- //上行流量
- String flowUp = split[2];
- PhoneFlowBean phoneFlowBean = new PhoneFlowBean(phone,Integer.parseInt(flowDown),Integer.parseInt(flowUp));
- //写入 K2 V2
- context.write(new Text(phone),phoneFlowBean);
- }
- }
复制代码- package com.ms.mshadoop.phoneFlow;
- import org.apache.hadoop.io.Text;
- import org.apache.hadoop.mapreduce.Reducer;
- import java.io.IOException;
- public class PhoneFlowReudcer extends Reducer<Text, PhoneFlowBean, Text, Text> {
- @Override
- protected void reduce(Text key, Iterable<PhoneFlowBean> values, Reducer<Text, PhoneFlowBean, Text, Text>.Context context) throws IOException, InterruptedException {
- //总下行流量
- int sumFlowDown = 0;
- //总上行流量
- int sumFlowUp = 0;
- for (PhoneFlowBean value : values) {
- sumFlowDown += value.getFlowDown();
- sumFlowUp += value.getFlowUp();
- }
- String flowInfo = String.format("总上行流量:%d,总下行流量:%d,总流量:%d。", sumFlowUp, sumFlowDown, (sumFlowUp + sumFlowDown));
- //写入 K3 V3
- context.write(key, new Text(flowInfo));
- }
- }
复制代码- package com.ms.mshadoop.phoneFlow;
- import org.apache.hadoop.conf.Configuration;
- import org.apache.hadoop.fs.Path;
- 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;
- public class PhoneFlowDirver {
- public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
- Configuration configuration = new Configuration();
- configuration.set("fs.defaultFS","file:///");
- configuration.set("mapreduce.framework.name","local");
- Job job = Job.getInstance(configuration);
- job.setMapperClass(PhoneFlowMapper.class);
- job.setReducerClass(PhoneFlowReudcer.class);
- job.setJarByClass(PhoneFlowDirver.class);
- job.setMapOutputKeyClass(Text.class);
- job.setMapOutputValueClass(PhoneFlowBean.class);
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(Text.class);
- //设置输入和输出的路径
- FileInputFormat.setInputPaths(job, new Path("输入文件路径"));
- //为防止文件已存在,可先进行判断,存在则删除文件
- FileOutputFormat.setOutputPath(job, new Path("输出文件路径"));
- System.exit( job.waitForCompletion(true)?0:1);
- }
- }
复制代码 免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |