大数据学习16_shuffle阶段&求共同好友项目实战

shuffle阶段

概念

shuffle 是 Mapreduce 的核心，它分布在 Mapreduce 的 map 阶段和 reduce 阶段。一般把从 Map 产生输出开始到 Reduce 取得数据作为输入之前的过程称作 shuffle。

一张图看懂Mapreduce全过程

概念解释

1. Collect阶段 ：将 MapTask 的结果输出到默认大小为 100M 的环形缓冲区，保存的是 key/value，Partition 分区信息等。
2. Spill阶段 ：当内存中的数据量达到一定的阀值的时候，就会将数据写入本地磁盘， 在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了 combiner，还会将 有相同分区号和 key 的数据进行排序。
3. Merge阶段 ：把所有溢出的临时文件进行一次合并操作，以确保一个 MapTask 最终只 产生一个中间数据文件。
4. Copy阶段 ：ReduceTask 启动 Fetcher 线程到已经完成 MapTask 的节点上复制一份属于 自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值 的时候，就会将数据写到磁盘之上。
5. Merge阶段 ：在 ReduceTask 远程复制数据的同时，会在后台开启两个线程对内存到本 地的数据文件进行合并操作。
6. Sort阶段 ：在对数据进行合并的同时，会进行排序操作，由于 MapTask 阶段已经对数 据进行了局部的排序，ReduceTask 只需保证 Copy 的数据的最终整体有效性即可。 Shule 中的缓冲区大小会影响到 mapreduce 程序的执行效率，原则上说，缓冲区越大， 磁盘io的次数越少，执行速度就越快 缓冲区的大小可以通过参数调整, 参数：mapreduce.task.io.sort.mb 默认100M

共同好友案例

需求分析

以下是qq的好友列表数据，冒号前是一个用户，冒号后是该用户的所有好友（数据中的好友 关系是单向的）

 求出哪些人两两之间有共同好友，及他俩的共同好友都有谁？

过程分析

mapreduce最重要的是根据自己的项目确定  K2和V2的类型，此案例中要用俩个mapreduce实现，将第一个mapreduce的输出作为第二个mapreduce的输入，第一个mapreduce要实现A-O分别是谁的好友，例如：A-E B；A-C-E D   B是A和E 的好友，D是A,C和E的好友。我们求的是俩俩之间的好友，要把A-C-E拆分成A-C A-E C-E；第二个阶段实现共同好友的求取。 

一个图看懂实现过程

代码实现

第一个mapper

package Commond_friends;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

//   K1行偏移量没有意义 ，V1是每行的内容
public class FirstMapper extends Mapper<LongWritable, Text,Text,Text> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:以冒号拆分行文本数据: 冒号左边就是V2
        String[] split = value.toString().split(":");
        String userStr = split[0];
        //2:将冒号右边的字符串以逗号拆分,每个成员就是K2
        String[] split1 = split[1].split(",");
        for (String s : split1) {
            //3:将K2和v2写入上下文中
            context.write(new Text(s),new Text(userStr));
        }
    }
}

第一个reduce

package Commond_friends;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class FirstReduce extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        //1:遍历集合,并将每一个元素拼接,得到K3
        StringBuffer buffer = new StringBuffer();
        for (Text value : values) {
            buffer.append(value.toString()+"-");
        }
        //2:K2就是V3
        //3:将K3和V3写入上下文中
        context.write(new Text(buffer.toString().substring(0,buffer.length()-1)),key);
    }
}

第一个jabmain

package Commond_friends;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class jobMain extends Configured implements Tool {
    @Override
    public int run(String[] strings) throws Exception {
        //1:获取Job对象
        Job job = Job.getInstance(super.getConf(), "common_friends_step1_job");

        //2:设置job任务
        //第一步:设置输入类和输入路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path("file:///D:\learnpath\Bigdata\project\Mapreduce\input\input.txt"));

        //第二步:设置Mapper类和数据类型
        job.setMapperClass(FirstMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //第三,四,五,六

        //第七步:设置Reducer类和数据类型
        job.setReducerClass(FirstReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //第八步:设置输出类和输出的路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path("file:///D:\learnpath\Bigdata\project\Mapreduce\input\input2"));

        //3:等待job任务结束
        boolean bl = job.waitForCompletion(true);



        return bl ? 0: 1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        //启动job任务
        int run = ToolRunner.run(configuration, new jobMain(), args);
        System.exit(run);
    }

}

第二个mapper

package common_friends_step2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Arrays;

public class scendMapper extends Mapper<LongWritable, Text,Text,Text> {
    /*
     K1           V1
     0           A-F-C-J-E- B
    ----------------------------------
     K2             V2
     A-C           B
     A-E           B
     A-F           B
     C-E           B
     */

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //1:拆分行文本数据,结果的第二部分可以得到V2
        String[] split = value.toString().split("	");
        String friend = split[1];

        //2:继续以'-'为分隔符拆分行文本数据第一部分,得到数组
        String[] userStr = split[0].split("-");
        Arrays.sort(userStr);
        //4:对数组中的元素进行两两组合,得到K2
        /*
          A-E-C -----> A C E
          A C E
            A C E
         */
        for(int i=0;i<userStr.length-1;i++)
        {
            for(int j=i+1;j<userStr.length;j++)
            {
                context.write(new Text(userStr[i]+"-"+userStr[j]),new Text(friend));
            }
        }


    }
}

第二个reduce

package common_friends_step2;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class secondReduce extends Reducer<Text,Text,Text,Text> {
    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        StringBuffer stringBuffer = new StringBuffer();
        for (Text value : values) {
            stringBuffer.append(value.toString()).append("-");
        }
        context.write(key,new Text(stringBuffer.toString().substring(0,stringBuffer.length()-1)));
    }
}

第二个jobmain

package common_friends_step2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class jobMain extends Configured implements Tool {
    @Override
    public int run(String[] strings) throws Exception {
        //1:获取Job对象
        Job job = Job.getInstance(super.getConf(), "common_friends_step2_job");

        //2:设置job任务
        //第一步:设置输入类和输入路径
        job.setInputFormatClass(TextInputFormat.class);
        TextInputFormat.addInputPath(job, new Path("file:///D:\learnpath\Bigdata\project\Mapreduce\input\input2"));

        //第二步:设置Mapper类和数据类型
        job.setMapperClass(scendMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        //第三,四,五,六

        //第七步:设置Reducer类和数据类型
        job.setReducerClass(secondReduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        //第八步:设置输出类和输出的路径
        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job, new Path("file:///D:\learnpath\Bigdata\project\Mapreduce\out"));

        //3:等待job任务结束
        boolean bl = job.waitForCompletion(true);



        return bl ? 0: 1;
    }

    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();

        //启动job任务
        int run = ToolRunner.run(configuration, new jobMain(), args);

        System.exit(run);
    }
}

 运行截图

初始文件表示一个用户的好友列表

 结果文件

示例：A-B  C-E  表示A和B 的共同好友有C和E