MapReduce概述（是一個做計算的程式）
MapReduce原理
Shuffle程序
MapReduce執行程序-map階段
MapReduce執行程序-reduce階段
shell端實作mapreduce
- 準備資料
- 實作mapreduce
- - 開始執行
  - 查看結果
Java操作mapreduce
Java操作mapreduce，指定reduce任務數
mapreduce程式關聯兩張表（實作join操作）
介于map端與reduce端之間的Combine
- mapreduce程式實作combinue預聚合

MapReduce概述（是一個做計算的程式）

MapReduce是一種分布式（一個計算邏輯，多個機器去實作）計算模型，由Google提出，主要用于搜索領域，解決海量資料的計算問題.
MapReduce是分布式運行的，由兩個階段組成：Map和Reduce，Map階段是一個獨立的程式，有很多個節點同時運行，每個節點處理一部分資料，Reduce階段是一個獨立的程式，有很多個節點同時運行，每個節點處理一部分資料【在這先把reduce理解為一個單獨的聚合程式即可】，
MapReduce框架都有默認實作，用戶只需要覆寫（重寫）map()和reduce()兩個函式，即可實作分布式計算，非常簡單，
這兩個函式的形參和回傳值都是<key、value>，使用的時候一定要注意構造<k,v>，

MapReduce原理

這里map階段和reduce階段的資料格式都有四個
map的前兩個保證了資料進來的格式，后兩個保證了資料出去的格式（也就是進到reduce的格式）
reduce的前兩個保證了資料進來的格式，后兩個保證了資料出去的格式（出去到hdfs中）
block塊的切分是hdfs切分的
split是mapreduce切分的（默認情況下，兩個的大小一致，都是128MB，允許溢位1.1）
一個split就對應這一個Mapper Task
在這里插入圖片描述
先拋開中間的shuffle不看，程序大致是這樣的：（這里面讀取資料用的是偏移量，是因為從hdfs讀取的是檔案，文本形式存盤的，key指的就是變數，value指的就是一行一行的資料）
先切分，
再傳到map階段，一個block塊對應一個map，
map里面先做一個簡單的計算，給需要做計數的單詞打上標記
在這里插入圖片描述

Shuffle程序

在這里插入圖片描述
在記憶體中做計算
map階段的shuffle
1.先從hdfs上讀取資料（hdfs自動切分，mapreduce切分獲取資料
每一個切片對應一個map task）

2.生成map task
在這里插入圖片描述

3.記憶體環形緩沖區（記憶體不可能都給用完，所以有一個默認大小100MB
100MB有一個溢位比例，達到閾值0.8，會溢寫到磁盤，會把80MB寫入
到磁盤，剩下的20MB在記憶體中做計算；剩下的20也不會一致在記憶體中；
當達到閾值0.8，或者map task計算完畢之后（計算完畢也沒必要在記憶體了）
都會溢寫到磁盤）
在這里插入圖片描述

4.三個就是磁區（達到閾值之后會再進行一個劃分）、
排序（按照當前資料某一類放在一起）、溢寫到磁盤
在這里插入圖片描述

5.三個箭頭匯總了：對第四步的小磁區檔案進行整合
在這里插入圖片描述
比方說這里有許多小磁區的資料，對這些小磁區的資料做一個聚合，方便后面的計算

6.發送，資料發送到reduce

reduce階段的shuffle
7.合并（將多個map task上的資料進行一個合并）
（可能有相同資料分發到同一個reduce，
合并之后再發送到reduce task，
資料來源可能在記憶體，可能在磁盤）（這時候的資料格式是hadoop 1）
在這里插入圖片描述

8.同一個key進入到同一個reduce；map階段有的資料可能在記憶體可能在磁盤，先做一個合并再發送
在這里插入圖片描述

在這里插入圖片描述

9.reduce合并計算結果（同一個key進入到一個reduce結果就出來了）
在這里插入圖片描述

10.合并結果放到hdfs
在這里插入圖片描述
匯總

MapReduce執行程序-map階段

1.1 框架使用InputFormat類的子類把輸入檔案(夾)劃分為很多InputSplit，默認，每個HDFS的block對應一個InputSplit，通過RecordReader類，把每個InputSplit決議成一個個<k1,v1>，默認，框架對每個InputSplit中的每一行，決議成一個<k1,v1>，
1.2 框架呼叫Mapper類中的map(…)函式，map函式的形參是<k1,v1>對，輸出是<k2,v2>對，一個InputSplit對應一個map task，程式員可以覆寫map函式，實作自己的邏輯，
1.3(假設reduce存在)框架對map輸出的<k2,v2>進行磁區，不同的磁區中的<k2,v2>由不同的reduce task處理，默認只有1個磁區，
(假設reduce不存在)框架對map結果直接輸出到HDFS中，
1.4 (假設reduce存在)框架對每個磁區中的資料，按照k2進行排序、分組，分組指的是相同k2的v2分成一個組，注意：分組不會減少<k2,v2>數量，
1.5 (假設reduce存在，可選)在map節點，框架可以執行reduce歸約，
1.6 (假設reduce存在)框架會對map task輸出的<k2,v2>寫入到linux 的磁盤檔案中，
至此，整個map階段結束

MapReduce執行程序-reduce階段

2.1 框架對多個map任務的輸出，按照不同的磁區，通過網路copy到不同的reduce節點，這個程序稱作shuffle，
2.2 框架對reduce端接收的[map任務輸出的]相同磁區的<k2,v2>資料進行合并、排序、分組，
2.3 框架呼叫Reducer類中的reduce方法，reduce方法的形參是<k2,{v2…}>，輸出是<k3,v3>，一個<k2,{v2…}>呼叫一次reduce函式，程式員可以覆寫reduce函式，實作自己的邏輯，
2.4 框架把reduce的輸出保存到HDFS中，

shell端實作mapreduce

準備資料

在這里插入圖片描述
由于mapreduce處理的資料應該來源于hdfs，所以在hdfs上創建資料傳入的路徑，也就是將剛才的資料傳到hdfs中

實作mapreduce

在這里插入圖片描述

值得注意的是這里的wordcount是自己取的名字，/word是hdfs已經存在的路徑，/output是hdfs中不存在的路徑

開始執行

在這里插入圖片描述

查看結果

在這里插入圖片描述

Java操作mapreduce

package com.shujia.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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 java.io.IOException;

public class Demo01WordCount {

    //統計檔案中單詞的個數
//重寫（覆寫）mapreduce中的map()和reduce()方法
//map類；第一對kv是決定資料輸入的格式，第二對kv決定資料輸出的格式
//第一個引數指的是偏移量，第二個引數指的是String型別，用的是文本形式
//后面兩個引數就是(hadoop,1)這樣的形式，最后一個引數用Integer也行
    public static class WCMapper extends Mapper<LongWritable,Text,Text,LongWritable> {
        /*
        map階段資料是一行一行過來的
        每一行資料都需要執行代碼
         */
        @Override
        //這里面三個引數，前面兩個引數可以理解為輸入的格式，第三個引數是輸出的資料(出去和進來的格式一模一樣)
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //通過context輸出Text  格式：（一整行資料，1）
            String line = value.toString();
            context.write(new Text(line),new LongWritable(1));
        }
    }

    //前兩個引數是map階段傳入的資料格式，后面兩個引數是輸出的格式
    public static class WCReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
        //這里面的值變成了一個迭代器(因為資料傳來的時候是【hadoop，1】，【hadoop，1】這樣的形式，但是只要一個hadoop就行了)
        //迭代器里面就是傳入的所有v的值
    /*
        reduce  聚合程式（有幾個key就跑幾次，意思就是檔案里面假如有一個hadoop，一個java，一個spark，那就需要跑三次）
        默認是一個節點
        key：每一個單詞
        values：map端 當前k所對應的所有v
         */
        //這里面要做的就是整合最后的結果(將迭代器中的資料相加)
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0L;
            for (LongWritable value : values) {
                sum+=value.get();
            }
            //把計算結果輸出到hdfs
            context.write(key,new LongWritable(sum));
        }
    }


    /*
    是當前mapreduce的入口
    用來構建mapreduce程式（資料從哪來等等）
     */
    public static void main(String[] args) throws Exception {
        Job job = Job.getInstance();
        //指定job名稱
        job.setJobName("單詞統計");
        //構建mr，指定當前main所在類名（識別具體的類）
        job.setJarByClass(Demo01WordCount.class);

        //指定map端類
        job.setMapperClass(WCMapper.class);
        //指定輸出的kv型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //指定reduce端類
        job.setReducerClass(WCReduce.class);
        //指定輸出的kv型別
        job.setReducerClass(WCReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        //指定輸入路徑
        Path in = new Path("/word");
        FileInputFormat.addInputPath(job,in);
        //指定輸出路徑
        Path out = new Path("/output");
        //如果路徑存在，洗掉
        FileSystem fs = FileSystem.get(new Configuration());
        if(fs.exists(out)){
            fs.delete(out,true);
        }
        FileOutputFormat.setOutputPath(job,out);

        //啟動任務
        job.waitForCompletion(true);
        System.out.println("mr正在執行");

        //hadoop jar hadoop-1.0-SNAPSHOT.jar com/shujia/hadoop/Demo01WordCount /word /output
    }
}

執行并獲得結果
在這里插入圖片描述

Java操作mapreduce，指定reduce任務數

在運行mr任務的時候，默認reduce執行一個，可以通過引數進行修改
job.setNumReduceTasks(2)

package com.shujia.hadoop;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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.yarn.logaggregation.AggregatedLogFormat;

import javax.security.auth.login.Configuration;
import java.io.FilterOutputStream;
import java.io.IOException;

public class Demo03SexSum {

    public static class WCMapper extends Mapper<LongWritable,Text,Text,LongWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] split = line.split(",");
            String sex = split[3];
            context.write(new Text(sex),new LongWritable(1));
        }
    }

    public static class WCReduce extends Reducer<Text,LongWritable,Text,LongWritable>{
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum=0L;
            for (LongWritable value : values) {
                sum+=value.get();
            }
            context.write(key,new LongWritable(sum));
        }
    }


    public static void main(String[] args) throws Exception{

        Job job = Job.getInstance();
        job.setNumReduceTasks(2);

        job.setJobName("統計不同性別人數");
        job.setJarByClass(Demo03SexSum.class);

        job.setMapperClass(WCMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        job.setReducerClass(WCReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        FileInputFormat.addInputPath(job,new Path("/word"));
        Path out = new Path("/out");
        FileSystem fs = FileSystem.get(new org.apache.hadoop.conf.Configuration());
        if(fs.exists(out)){
            fs.delete(out,true);
        }
        FileOutputFormat.setOutputPath(job,out);

        job.waitForCompletion(true);

        //com.shujia.hadoop.Demo03SexSum
    }
}

mapreduce程式關聯兩張表（實作join操作）

package com.shujia.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.InputSplit;
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.input.FileSplit;
//import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;

/*
students.txt
1500100001,施笑槐,22,女,文科六班
1500100002,呂金鵬,24,男,文科六班
1500100003,單樂蕊,22,女,理科六班
1500100004,葛德曜,24,男,理科三班
1500100005,宣谷芹,22,女,理科五班

score.txt
1500100001,1000001,98
1500100001,1000002,5
1500100001,1000003,137
1500100001,1000004,29
1500100001,1000005,85
 */
public class Demo05Join {

    public static class JoinMapper extends Mapper<LongWritable,Text,Text,Text>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //1.獲取資料的路徑InputSplit
            //context上面是hdfs，下面如果有reduce就是reduce，沒有就是hdfs
            InputSplit inputSplit = context.getInputSplit();
            FileSplit fs = (FileSplit)inputSplit;
            String url = fs.getPath().toString();
            //2.判斷
            if (url.contains("students")){//true：當前的資料為students.txt
                String id = value.toString().split(",")[0];
                //為了方便reduce資料的操作，針對不同的資料 打上一個標簽
                String line = "*"+value.toString();
                context.write(new Text(id),new Text(line));
            }else {
                //學號作為key，是兩張表的關聯條件
                String id = value.toString().split(",")[0];
                String line = "#"+value.toString();
                context.write(new Text(id),new Text(line));
            }
        }
    }

    public static class JoinReduce extends Reducer<Text,Text,Text,NullWritable>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            ArrayList<String> scores = new ArrayList<String>();
            String stuinfo = "";
            for (Text value : values) {
                String line = value.toString();
                if (line.startsWith("*")){
                    stuinfo = line.substring(1);
                }else {
                    scores.add(line.substring(1));
                }
            }

            //資料拼接，兩張表的拼接
//            for (String score : scores) {
//                String subject = score.split(",")[1];
//                String sc = score.split(",")[2];
//                String info = stuinfo+","+subject+","+sc;
//                context.write(new Text(info),NullWritable.get());
//            }

            //成績求和，兩張表的拼接
            long sum = 0L;
            for (String score : scores) {
                Integer sc = Integer.parseInt(score.split(",")[2]);
                sum+=sc;
            }
            String end=stuinfo+","+sum;
            context.write(new Text(end),NullWritable.get());

        }
    }


    public static void main(String[] args) throws Exception{

        Job job = Job.getInstance();
        job.setJarByClass(Demo05Join.class);
        job.setJobName("Join操作");

        job.setMapperClass(JoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(JoinReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //路徑
        FileInputFormat.addInputPath(job,new Path("/word"));
        Path out = new Path("/out");
        FileSystem fs = FileSystem.get(new Configuration());
        if (fs.exists(out)){
            fs.delete(out,true);
        }
        FileOutputFormat.setOutputPath(job,out);
        job.waitForCompletion(true);
        System.out.println("程式正在執行");

    }
}

介于map端與reduce端之間的Combine

combiner發生在map端的reduce操作，
作用是減少map端的輸出，減少shuffle程序中網路傳輸的資料量，提高作業的執行效率，
combiner僅僅是單個map task的reduce，沒有對全部map的輸出做reduce，
如果不用combiner，那么，所有的結果都是reduce完成，效率會相對低下，使用combiner，先完成的map會在本地聚合，提升速度，
注意：Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結果，所以，Combine適合于等冪操作，比如累加，最大值等，求平均數不適合

如圖，中間的這一部分就是combine，提高效率一眼可見
在這里插入圖片描述

mapreduce程式實作combinue預聚合

package com.shujia.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
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 java.io.IOException;

//LongWritable 偏移量 long，表示該行在檔案中的位置
public class Demo06Combine {
    public static class CombineMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String sex = line.split(",")[3];
            context.write(new Text(sex),new IntWritable(1));
        }
    }

    //預聚合在reduce之前 map端之后
    public static class Combine extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            Integer sum=0;
            for (IntWritable value : values) {
                sum+=value.get();
            }
            context.write(key,new IntWritable(sum));
        }
    }

    public static class CombineReduce extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            Integer sum=0;
            for (IntWritable value : values) {
                sum+=value.get();
            }
            context.write(key,new IntWritable(sum));
        }
    }

    public static void main(String[] args) throws Exception{
        Job job = Job.getInstance();
        job.setNumReduceTasks(2);
        job.setJobName("假如Combine做性別統計");
        job.setJarByClass(Demo06Combine.class);
        job.setMapperClass(CombineMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setReducerClass(CombineReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job,new Path("/word"));
        Path path = new Path("/out");
        FileSystem fs = FileSystem.get(new Configuration());
        if (fs.exists(path)){
            fs.delete(path,true);
        }
        FileOutputFormat.setOutputPath(job,path);
        job.waitForCompletion(true);


    }
}

感謝閱讀，我是啊帥和和，一位大資料專業大四學生，祝你快樂，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/302461.html

標籤：其他

上一篇：基于Echarts+HTML5可視化資料大屏展示—電子商務公共服務平臺大資料中心

下一篇：一篇入門分布式事務

深入理解MapReduce

目錄

MapReduce概述（是一個做計算的程式）

MapReduce原理

Shuffle程序

MapReduce執行程序-map階段

MapReduce執行程序-reduce階段

shell端實作mapreduce

準備資料

實作mapreduce

開始執行

查看結果

Java操作mapreduce

Java操作mapreduce，指定reduce任務數

mapreduce程式關聯兩張表（實作join操作）

介于map端與reduce端之間的Combine

mapreduce程式實作combinue預聚合