Hadoop-MapReduce

小小小幸运  金牌会员 | 2024-7-23 15:57:25 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 504|帖子 504|积分 1512

MapReduce

1、优缺点

  1. 优点：易于编程、良好的扩展性、高容错性
  2. 缺点：不适合做实时计算、不适合做流式计算(要求数据是静态的)、不适合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
  1. package com.ms.mshadoop.mapreduce;
  2. import org.apache.hadoop.io.IntWritable;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  9. /**
  10. * Map阶段开始的时候，每一个MapTask都会逐行读取分片中的数据，并将读取到的数据进行扭转形成 keyIn, valueIn
  11. * 由于计算程序需要到不同节点之间进行移动，因此涉及到 keyIn/valueIn/keyOut/valueOut都必须支持序列化的类型
  12. * Hadoop提供了一套序列化的机制 writeable
  13. * byte    =>  ByteWritable
  14. * short   =>  ShortWritable
  15. * int     =>  IntWritable
  16. * long    =>  LongWritable
  17. * float   =>  FloatWritable
  18. * double  =>  DoubleWritable
  19. * boolean =>  BooleanWritable
  20. * String  =>  Text
  21. * <p>
  22. * keyIn:读取到的 行数据 中 首字母 的偏移量  需要设计为LongWritable
  23. * valueIn:读取到的 行数据                需要设计为Text
  24. * <p>
  25. * keyOut:经过逻辑处理后，需要写出的键值对中 键的类型 Text
  26. * valueOut:经过逻辑处理后，需要写出的键值对中 值的类型 IntWritable
  27. */
  28. public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
  30.     /**
  31.      * 每当读取到一行数据时，将其扭转为keyIn和valueIn,调用该方法
  32.      * @param key  行偏移量
  33.      * @param value 行记录
  34.      * @param context 操作上下文
  35.      * @throws IOException
  36.      * @throws InterruptedException
  37.      */
  38.     @Override
  39.     protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
  40.         //1、将读取的行数据，切割出每一个单词
  41.         // 正则表达式中\s匹配任何空白字符，包括空格、制表符、换页符等等, 等价于[\f\n\r\t\v]
  42.         // 而\s+则表示匹配任意多个上面的字符
  43.         String[] words = value.toString().split("\\s+");
  44.         //2、遍历每一个单词
  45.         for (String word : words) {
  46.             //3、为每一单词配上(value,1) 组成键值对 写出
  47.             context.write(new Text(word),new IntWritable(1));
  48.         }
  49.     }
  50. }
复制代码
  1. package com.ms.mshadoop.mapreduce;
  2. import org.apache.hadoop.io.IntWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  8. /**
  9. * keyIn: Map阶段输出的键类型
  10. * valueIn: Map阶段输出的值类型
  11. * <p>
  12. * keyOut: 最终输出的键值对中，键的类型
  13. * valueOut: 最终输出的键值对中，值的类型
  14. */
  15. public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
  17.     /**
  18.      * Reduce阶段逻辑处理
  19.      *
  20.      * @param key     键，Map阶段输出的键
  21.      * @param values  输入进这个方法之前，MapReduce会按照键进行分组，将相同的键对应的所有值聚合到一起
  22.      * @param context 上下文
  23.      * @throws IOException
  24.      * @throws InterruptedException
  25.      */
  26.     @Override
  27.     protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
  28.         //1、定义一个变量，来记录单子出现的总次数
  29.         int timeCount = 0;
  30.         for (IntWritable value : values) {
  31.             timeCount += value.get();
  32.         }
  33.         context.write(key, new IntWritable(timeCount));
  34.     }
  35. }
复制代码
  1. package com.ms.mshadoop.mapreduce;
  2. import org.apache.hadoop.conf.Configuration;
  3. import org.apache.hadoop.fs.Path;
  4. import org.apache.hadoop.io.IntWritable;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. import java.io.IOException;
  12. /**
  13. * 驱动类:主要功能
  14. * 1、创建Job
  15. * 2、Job属性配置
  16. * 3、Job提交
  17. */
  18. public class WordCountDriver {
  20.     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
  21.         //1、创建Job
  22.         Configuration conf = new Configuration();
  23.         conf.set("fs.defaultFS", "hdfs://hadoop01:9820");
  24.         Job job = Job.getInstance();
  26.         //2、设置Job属性
  27.         //在MapReduce程序中，用于处理Map任务的类
  28.         job.setMapperClass(WordCountMapper.class);
  29.         //在MapReduce程序中，用于处理Reduce任务的类
  30.         job.setReducerClass(WordCountReducer.class);
  31.         //在MapReduce程序中，用于处理驱动的类
  32.         job.setJarByClass(WordCountDriver.class);
  33.         //Map阶段输出的键值对的类型
  34.         job.setMapOutputKeyClass(Text.class);
  35.         job.setMapOutputValueClass(IntWritable.class);
  36.         //Reduce阶段输出的键值对的类型
  37.         job.setOutputKeyClass(Text.class);
  38.         job.setOutputValueClass(IntWritable.class);
  40.         //设置输入和输出的路径
  41.         FileInputFormat.setInputPaths(job, new Path(args[0]));
  42.         FileOutputFormat.setOutputPath(job, new Path(args[1]));
  44.         //3、提交任务
  45.         System.exit(job.waitForCompletion(true) ? 0 : -1);
  46.     }
  47. }
复制代码
注意：需要天生多个文件数量时，代码中增长
  1. //设置ReduceTask的数量,决定了最终生成的文件数量。这个数量最好和分区的数量保持一致。
  2.         /**
  3.          * 1、如果ReduceTask的数量多于分区的数量:会出现多余的ReduceTask空占资源，
  4.          *      不去处理任何的数据，浪费资源，且生成空的结果文件
  5.          * 2、如果ReduceTask的数量少于分区的数量:会出现某个分区的数据暂时无法处理，
  6.          *      需要等待某ReduceTask任务处理结束后再1处理这个分区的数据，无法高效并发
  7.          */
  8. job.setNumReduceTasks(2);
复制代码
3.2 执行步调

  1. 1、IDE中将项目打包成jar，上传到linux 的hadoop集群中
  2. 2、运行程序
  3. # hadoop jar 打包上传到服务器的jar  程序中WordCountDriver的包路径         /输入文件路径        /输出文件路径
  4. 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相同

  1. package com.ms.mshadoop.partitioner;
  2. import org.apache.hadoop.io.IntWritable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Partitioner;
  6. /**
  7. * Partitioner : 分区器，可以将map阶段输出的键值对，按照规则进行分区
  8. * 泛型，对应的是map方法输出的键值对类型
  9. */
  10. public class WordCountPartitioner extends Partitioner<Text, IntWritable> {
  11.     /**
  12.      * 每一个map方法输出的键值对，都会调用该方法进行分区，来确定分区号
  13.      * @param text map 阶段输出的 键
  14.      * @param intWritable map 阶段输出的 值
  15.      * @param i ReduceTask 的数量
  16.      * @return 分区号，从零开始，且必须连续
  17.      */
  18.     @Override
  19.     public int getPartition(Text text, IntWritable intWritable, int i) {
  20.         char c = text.toString().charAt(0);
  21.         if (c >= 'a' && c <= 'g') {
  22.             return 0;
  23.         } else if (c >= 'h' && c <= 'o') {
  24.             return 1;
  25.         } else {
  26.             return 2;
  27.         }
  28.     }
  29. }
复制代码
WordCountDriver中增长如下代码
  1.                 //应用分区器
  2.         job.setPartitionerClass(WordCountPartitioner.class);
  4.         //设置ReduceTask的数量,决定了最终生成的文件数量。这个数量最好和分区的数量保持一致。
  5.         /**
  6.          * 1、如果ReduceTask的数量多于分区的数量:会出现多余的ReduceTask空占资源，
  7.          *      不去处理任何的数据，浪费资源，且生成空的结果文件
  8.          * 2、如果ReduceTask的数量少于分区的数量:会出现某个分区的数据暂时无法处理，
  9.          *      需要等待某ReduceTask任务处理结束后再1处理这个分区的数据，无法高效并发
  10.          */
  11.         job.setNumReduceTasks(3);
复制代码
5、IDE运行MapReduce的模式

  1. 在IDEA中运行MapReduce的程序，可以选择计算资源，也可以选择文件系统
  2.     计算资源：mapreduce.framework.name
  3.                 local:使用本地计算资源（CPU、内存）
  4.                 yarn:使用集群的计算资源
  5.     文件系统：fs.defaultFS
  6.                 hdfs://hadoop01:9820    使用分布式文件系统
  7.                 file:///                使用本地文件系统
  8. 本地调试时可使用local模式进行，正常推荐使用yarn模式进行
复制代码
5.1、local 模式测本地文件

原理：
1、将MapReduce的任务资源调度设置为local，倒霉用YARN举行资源调度
2、将文件系统设置为本地文件系统，倒霉用HDFS
  1. Configuration conf = new Configuration();
  2. conf.set("mapreduce.framework.name","local");//设置为本地运行模式，任务不会在YARN上运行
  3. conf.set("fs.defaultFS","file:///");//设置为本地文件系统，不使用HDFS。
  5. Job job = Job.getInstance(conf );
  6. //设置输入和输出的路径
  7. FileInputFormat.setInputPaths(job, new Path("/输入：本地文件路径"));
  8. FileOutputFormat.setOutputPath(job, new Path("/输出：本地文件路径"));
复制代码
5.2、local 模式测集群文件

原理：
1、将MapReduce的任务资源调度设置为local，倒霉用YARN举行资源调度。
2、将文件系统设置为HDFS
  1. System.setProperty("HADOOP_USER_NAME", "root");        //设置hadoop的操作用户
  2. Configuration conf = new Configuration();
  3. conf.set("mapreduce.framework.name","local");//设置为本地运行模式，任务不会在YARN上运行
  4. conf.set("fs.defaultFS","hdfs://hadoop01:9820");//设置为分布式文件系统
  5. Job job = Job.getInstance(conf );
  7. //设置输入和输出的路径
  8. FileInputFormat.setInputPaths(job, new Path("/输入：hdfs文件路径"));
  9. FileOutputFormat.setOutputPath(job, new Path("/输出：dhfs文件路径"));
复制代码
5.3、YARN模式测集群

原理：
1、将MapReduce的任务资源调度设置为YARN
2、将文件系统设置为HDFS
  2.         System.setProperty("HADOOP_USER_NAME", "root");        //设置hadoop的操作用户
  3.         //1、创建Job
  4.         Configuration conf = new Configuration();
  5.         //local 模式测本地文件
  6.         /*conf.set("mapreduce.framework.name","local");//设置为本地运行模式，任务不会在YARN上运行
  7.         conf.set("fs.defaultFS","file:///");//设置为本地文件系统，不使用HDFS。*/
  8.         //local 模式测集群文件
  9.         /*conf.set("mapreduce.framework.name","local");//设置为本地运行模式，任务不会在YARN上运行
  10.         conf.set("fs.defaultFS","hdfs://hadoop01:9820");//设置为分布式文件系统*/
  12.         //yarn 模式测集群文件
  13.         conf.set("mapreduce.framework.name", "yarn");//设置为本地运行模式，任务不会在YARN上运行
  14.         conf.set("yarn.resourcemanager.hostname", "hadoop01");//设置ResourceManager
  15.         conf.set("fs.defaultFS", "hdfs://hadoop01:9820");//设置为分布式文件系统
  16.         conf.set("yarn.app.mapreduce.am.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的ApplicationMaster（AM）启动时的环境变量。
  17.         conf.set("mapreduce.map.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的Map启动时的环境变量。
  18.         conf.set("mapreduce.reduce.env", "HADOOP_MAPRED_HOME=/usr/local/hadoop-3.3.6");//用于指定MapReduce应用程序的Reduce启动时的环境变量。
  19.         conf.set("hadoop.security.authentication","simple");
  20.         conf.set("mapreduce.app-submission.cross-platform","true");//跨平台提交任务
复制代码
注意事项:
在上述的配置都完成后，将步调打jar包，然后将jar包添加到classpath，才可以运行步调
  1. 在jar包上右键-> Add As Library
复制代码
6 Hadoop 序列化

  1. package com.ms.mshadoop.writable;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.io.Writable;
  6. import java.io.DataInput;
  7. import java.io.DataOutput;
  8. import java.io.IOException;
  9. import java.io.Serializable;
  11. public class Student implements Writable, Serializable {
  13.     private String name;
  14.     private int age;
  15. //    private Text remarks;
  18.     public Student() {
  19.     }
  21.     public Student(String name, int age, Text remarks) {
  22.         this.name = name;
  23.         this.age = age;
  24. //        this.remarks = remarks;
  25.     }
  27.     /**
  28.      * 序列化：将内存对象，序列化为一个字节序列
  29.      * @param dataOutput
  30.      */
  31.     @Override
  32.     public void write(DataOutput dataOutput) throws IOException {
  33.         //该方法中依次将所有属性序列化
  35.         //当属性非hadoop中的序列类型时
  36.         //将String 序列化
  37.         dataOutput.writeUTF(name);
  38.         //将int 序列化
  39.         dataOutput.writeInt(age);
  41.         //当属性为hadoop中的序列化类型时
  42. //        remarks.write(dataOutput);
  44.     }
  46.     /**
  47.      * 反序列化：将字节序列转 内存对象
  48.      * 注意：反序列化时，读取属性顺序 必须 和序列化时顺序一致
  49.      * @param dataInput
  50.      */
  51.     @Override
  52.     public void readFields(DataInput dataInput) throws IOException {
  53.         //当属性非hadoop中的序列类型时
  54.         //将String 反序列化
  55.         name = dataInput.readUTF();
  56.         //将int 反序列化
  57.         age = dataInput.readInt();
  58.         //当属性为hadoop中的序列化类型时
  59. //        remarks.readFields(dataInput);
  60.     }
  61. }
  64. package com.ms.mshadoop.writable;
  66. import org.apache.hadoop.io.Text;
  68. import java.io.*;
  70. public class Test {
  71.     public static void main(String[] args) throws IOException {
  73.         Student student = new Student("张三丰",100,new Text("11111111"));
  74.         File file = new File("C:\\Users\\xazyh\\Desktop\\1111.txt");
  75.         ObjectOutputStream outputStream = new ObjectOutputStream(new FileOutputStream(file));
  76.         outputStream.writeObject(student);
  77.         outputStream.close();
  78.         File file1= new File("C:\\Users\\xazyh\\Desktop\\2222.txt");
  79.         DataOutputStream dataOutputStream = new DataOutputStream(new FileOutputStream(file1));
  80.         student.write(dataOutputStream);
  81.         dataOutputStream.close();
  82.     }
  83. }
复制代码

7 Mapreduce 实现统计流量

  1. package com.ms.mshadoop.phoneFlow;
  3. import org.apache.hadoop.io.Writable;
  5. import java.io.DataInput;
  6. import java.io.DataOutput;
  7. import java.io.IOException;
  9. public class PhoneFlowBean implements Writable {
  11.     private String phone;
  13.     private int flowDown;
  15.     private int flowUp;
  17.     public PhoneFlowBean() {
  18.     }
  20.     public PhoneFlowBean(String phone, int flowDown, int flowUp) {
  21.         this.phone = phone;
  22.         this.flowDown = flowDown;
  23.         this.flowUp = flowUp;
  24.     }
  26.     @Override
  27.     public void write(DataOutput dataOutput) throws IOException {
  29.         dataOutput.writeUTF(phone);
  30.         dataOutput.writeInt(flowDown);
  31.         dataOutput.writeInt(flowUp);
  33.     }
  35.     @Override
  36.     public void readFields(DataInput dataInput) throws IOException {
  38.         phone = dataInput.readUTF();
  39.         flowDown = dataInput.readInt();
  40.         flowUp = dataInput.readInt();
  42.     }
  44.     public int getSumFlow() {
  45.         return flowDown + flowUp;
  46.     }
  48.     public String getPhone() {
  49.         return phone;
  50.     }
  52.     public void setPhone(String phone) {
  53.         this.phone = phone;
  54.     }
  56.     public int getFlowDown() {
  57.         return flowDown;
  58.     }
  60.     public void setFlowDown(int flowDown) {
  61.         this.flowDown = flowDown;
  62.     }
  64.     public int getFlowUp() {
  65.         return flowUp;
  66.     }
  68.     public void setFlowUp(int flowUp) {
  69.         this.flowUp = flowUp;
  70.     }
  71. }
复制代码
  1. package com.ms.mshadoop.phoneFlow;
  3. import org.apache.hadoop.io.LongWritable;
  4. import org.apache.hadoop.io.Text;
  5. import org.apache.hadoop.mapreduce.Mapper;
  7. import java.io.IOException;
  9. public class PhoneFlowMapper extends Mapper<LongWritable, Text, Text, PhoneFlowBean> {
  11.     @Override
  12.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  13.         String line = value.toString();
  14.         //根据数据格式进行数据分割
  15.         String[] split = line.split("\t");
  16.         //手机号
  17.         String phone = split[0];
  18.         //下行流量
  19.         String flowDown = split[1];
  20.         //上行流量
  21.         String flowUp = split[2];
  23.         PhoneFlowBean phoneFlowBean = new PhoneFlowBean(phone,Integer.parseInt(flowDown),Integer.parseInt(flowUp));
  24.         //写入 K2 V2
  25.         context.write(new Text(phone),phoneFlowBean);
  26.     }
  27. }
复制代码
  1. package com.ms.mshadoop.phoneFlow;
  3. import org.apache.hadoop.io.Text;
  4. import org.apache.hadoop.mapreduce.Reducer;
  6. import java.io.IOException;
  8. public class PhoneFlowReudcer extends Reducer<Text, PhoneFlowBean, Text, Text> {
  10.     @Override
  11.     protected void reduce(Text key, Iterable<PhoneFlowBean> values, Reducer<Text, PhoneFlowBean, Text, Text>.Context context) throws IOException, InterruptedException {
  12.         //总下行流量
  13.         int sumFlowDown = 0;
  14.         //总上行流量
  15.         int sumFlowUp = 0;
  16.         for (PhoneFlowBean value : values) {
  17.             sumFlowDown += value.getFlowDown();
  18.             sumFlowUp += value.getFlowUp();
  19.         }
  21.         String flowInfo = String.format("总上行流量：%d，总下行流量：%d，总流量：%d。", sumFlowUp, sumFlowDown, (sumFlowUp + sumFlowDown));
  22.         //写入 K3 V3
  23.         context.write(key, new Text(flowInfo));
  24.     }
  27. }
复制代码
  1. package com.ms.mshadoop.phoneFlow;
  3. import org.apache.hadoop.conf.Configuration;
  4. import org.apache.hadoop.fs.Path;
  5. import org.apache.hadoop.io.Text;
  6. import org.apache.hadoop.mapreduce.Job;
  7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
  8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
  10. import java.io.IOException;
  12. public class PhoneFlowDirver {
  14.     public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
  15.         Configuration configuration = new Configuration();
  16.         configuration.set("fs.defaultFS","file:///");
  17.         configuration.set("mapreduce.framework.name","local");
  19.         Job job = Job.getInstance(configuration);
  21.         job.setMapperClass(PhoneFlowMapper.class);
  22.         job.setReducerClass(PhoneFlowReudcer.class);
  23.         job.setJarByClass(PhoneFlowDirver.class);
  25.         job.setMapOutputKeyClass(Text.class);
  26.         job.setMapOutputValueClass(PhoneFlowBean.class);
  28.         job.setOutputKeyClass(Text.class);
  29.         job.setOutputValueClass(Text.class);
  31.         //设置输入和输出的路径
  32.         FileInputFormat.setInputPaths(job, new Path("输入文件路径"));
  33.         //为防止文件已存在，可先进行判断，存在则删除文件
  34.         FileOutputFormat.setOutputPath(job, new Path("输出文件路径"));
  36.        System.exit( job.waitForCompletion(true)?0:1);
  37.     }
  38. }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

小小小幸运

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表