MapReduce的Shuffle機制
1、MapReduce的shuffle機制
1.1、概述
MapReduce中，mapper階段處理的資料如何傳遞給reduce階段，是MapReduce框架中最關鍵的一個流程，這個流程就叫shuffle.
Shuffle:資料混洗---------（核心機制：資料分割槽，排序，區域性聚合，快取，拉取，再合併排序）
具體來說，就是將MapTask輸出的處理資料結果，按照Partitioner元件制定的規則分發ReduceTask，並在分發的過程中，對資料按key進行分割槽和排序
1.2、主要流程
Shuffle是MapReduce處理流程中的一個核心，它的每一個處理步驟是分散在各個Maptask和reducetask節點上完成的，整體來看，分為3個操作：
1、分割槽partition(如果reduceTask只有一個或者沒有，那麼partition將不起作用。設定沒設定相當於沒有)
2、Sort根據key排序（MapReduce程式設計中sort是一定會做的，並且只能按照key排序，當然如果沒有reduce階段，那麼就不會對key排序）
3、Combiner進行區域性value的合併（Combiner是可選的元件，作用是為了提高任務的執行效率）
1.3、詳細流程
1、mapTask收集我們map()方法輸出的kv對，放在記憶體緩衝區kvbuffer（環形緩衝區：記憶體中的一種首尾相連的資料結構，kvbuffer包含資料區和索引區）中，在存資料的時候，會呼叫partitioner進行分割槽編號的計算，並存入元資料中
2、當記憶體緩衝區的資料達到100*0.8時，就會開始溢寫到本地磁碟檔案file.out，可能會溢位多次，則會有多個檔案，相應的緩衝區中的索引區資料溢位為磁碟索引檔案file.out.index
3、在溢寫前，會先根據分割槽編號排序，相同的分割槽的資料，排在一起，再根據map的key排序（快排）
4、多個溢寫檔案會被合併成大的溢位檔案（歸併排序）
5、在資料量大的時.候，可以對maptask結果啟用壓縮，將mapreduce.map.ouput.compress設為true，並使用
mapreduce.map.output.compress.codec設定使用的壓縮演算法，可以提高資料傳輸到reduce端的效率
6、reduceTask根據自己的分割槽號，去各個mapTask機器上取相應的結果分割槽資料
7、reduceTask會取到同一個分割槽的來自不同mapTask的結果檔案，reduceTask會將這些檔案再進行合併（歸併排序）
8、合併成r大檔案後，shuffle的過程也就結束了，後面進入reduceTask的邏輯運算過程（從檔案中取出一個一個的鍵值對group，呼叫使用者自定義的reduce(）方法）
2、自定義Shuffle過程中的元件
1、自定義輸入
預設輸入類：TextInputFormat
自定義：
模仿 org.apache.hadoop.mapreduce.lib.input.LineRecordReader 和org.apache.hadoop.mapreduce.lib.input.TextInputFormat
1、自定義類繼承FileInputFormat
public class MyFileInputFormat extends FileInputFormat<Text, LongWritable>{
@Override
public RecordReader<Text, LongWritable> createRecordReader(InputSplit split, TaskAttemptContext context)
throws IOException, InterruptedException {
//例項化一個
MyAllFileRecodReader reader = new MyAllFileRecodReader();
//split引數和context都是框架自動傳入的，把這兩個引數傳給reader進行處理，以便獲取相關資訊
reader.initialize(split, context);
return reader;
}
/**
* 給定的檔名可拆分嗎?返回false確保單個輸入檔案不會被分割。以便Mapper處理整個檔案。
*/
@Override
protected boolean isSplitable(JobContext context, Path filename) {
return false;
}
}
2、自定義類實現RecordReader
public class MyFileRecodReader extends RecordReader<Text, LongWritable>{
//用於儲存檔案系統輸入流
private FSDataInputStream open = null;
//儲存檔案長度
private int fileSplitLength = 0;
/**
* 當前的MyAllFileRecodReader讀取到的一個key-value
*/
private Text key = new Text();
private LongWritable value = new LongWritable();
@Override
public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
//通過InputSplit物件獲取檔案路徑
FileSplit fileSplit = (FileSplit)split;
Path path = fileSplit.getPath();
//獲取檔案長度
fileSplitLength = (int)fileSplit.getLength();
//通過context物件獲取到配置檔案資訊，通過配置檔案獲取到一個當前檔案系統
Configuration configuration = context.getConfiguration();
FileSystem fs = FileSystem.get(configuration);
//獲取檔案系統的一個輸入流
open = fs.open(path);
}
/**
* 已讀標記
* 如果為false，表示還沒有進行讀取
* 在需求中一個mapTask只處理一個小檔案，一個mapTask最終只需要讀取一次就完畢
* 如果一個檔案讀取完畢了，那麼就把isRead這個變數標記為true
*/
private boolean isRead = false;
/**
* 實現讀取規則：逐檔案讀取
*/
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
//如果沒有讀取過檔案就進入
if(!isRead){
//準備一個位元組陣列長度為檔案的長度
byte[] buffer = new byte[fileSplitLength];
//一次性把真個檔案讀入位元組陣列中
IOUtils.readFully(open, buffer);
//把讀取到的檔案傳給key
key.set(buffer, 0, fileSplitLength);
//設定已讀標記為true
isRead = true;
//返回讀取一個檔案成功標記
return true;
}else{
return false;
}
}
//獲取key的方法
@Override
public Text getCurrentKey() throws IOException, InterruptedException {
return key;
}
//獲取當前value值
@Override
public LongWritable getCurrentValue() throws IOException, InterruptedException {
return value;
}
/**
* 獲取資料的處理進度的
*/
@Override
public float getProgress() throws IOException, InterruptedException {
//已讀為真返回1.0，沒有讀返回0
return isRead ? 1.0F : 0F;
}
@Override
public void close() throws IOException {
//關閉輸入流
IOUtils.closeQuietly(open);
}2、自定義分割槽
需要： 1、繼承 partitioner
2、重寫getpartition()方法
3、在main方法中指定分割槽類 job.setPartitionclass()
package homework;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class Mypartition extends Partitioner<Student, Text> {
@Override
public int getPartition(Student key, Text arg1, int arg2) {
if(key.getType().equals("math")){
return 0;
}
if(key.getType().equals("english")){
return 1;
}
if(key.getType().equals("computer")){
return 2;
}else{
return 3;
}
}
}3、自定義排序
需要 ： 1、實現writableComparable
2、重新write()、readFields()、compareTo()方法
package homework;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
public class Student implements WritableComparable<Student> {
private String type;
private String name;
private Double avg;
public Student() {
super();
}
public Student(String type, String name, Double avg) {
super();
this.type = type;
this.name = name;
this.avg = avg;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Double getAvg() {
return avg;
}
public void setAvg(Double avg) {
this.avg = avg;
}
@Override
public String toString() {
return type + "\t" + name + "\t" + avg ;
}
@Override
public void readFields(DataInput in) throws IOException {
this.type=in.readUTF();
this.name=in.readUTF();
this.avg=in.readDouble();
}
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(type);
out.writeUTF(name);
out.writeDouble(avg);
}
@Override
public int compareTo(Student o) {
int temp=o.getType().compareTo(this.getType());
if(temp==0){
if(o.getAvg()>this.getAvg()){
return 1;
}else if(o.getAvg()<this.getAvg()){
return -1;
}else{
return 0;
}
}
return temp;
}
}
4、自定義分組
需要 ： 1、繼承writableComparable
2、重寫compare()方法
3、指定分組類 job.setGroupingComparatorClass(MyGroup.class);
4、既有分割槽又有排序的時候，分組欄位一定在排序欄位中
package homework;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class MyGroup extends WritableComparator {
public MyGroup() {
super(Student.class,true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
Student aa=(Student)a;
Student bb=(Student)b;
return aa.getType().compareTo(bb.getType());
}
}
5、自定義輸出
1）模仿 org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
public class MyMultipePathOutputFormat extends FileOutputFormat<Text, NullWritable>{
@Override
public RecordWriter<Text, NullWritable> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
//獲得當前的檔案系統傳給自定義的RecordWriter元件
Configuration configuration = job.getConfiguration();
FileSystem fs = FileSystem.get(configuration);
try {
//返回一個RecordWriter正在處理輸出資料的元件
return new MyMutiplePathRecordWriter(fs);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
2）繼承RecordWriter 並實現write()方法
public class MyMutiplePathRecordWriter extends RecordWriter<Text, NullWritable>{
//宣告要輸出的兩個路徑
private DataOutputStream out_jige;
private DataOutputStream out_bujige;
public MyMutiplePathRecordWriter(FileSystem fs) throws Exception {
//建立系統輸出流
out_jige = fs.create(new Path("E:\\bigdata\\cs\\jige\\my_output_jige.txt"));
out_bujige = fs.create(new Path("E:\\bigdata\\cs\\bujige\\my_output_bujige.txt"));
}
/**
* 實現寫出方法，根據需要寫出的格式自定義
*/
@Override
public void write(Text key, NullWritable value) throws IOException, InterruptedException {
//接受到的key格式為：course + "\t" + name + "\t" + avgScore
String keyStr = key.toString();
String[] split = keyStr.split("\t");
//獲取到平均分欄位
double score = Double.parseDouble(split[2]);
//沒一行資料加入個換行符
byte[] bytes = (keyStr + "\n").getBytes();
//如果平均分大於60就用DataOutputStream寫出到jige目錄
if(score >= 60){
out_jige.write(bytes, 0, bytes.length);
}else{//小於60分的寫道bujige目錄
out_bujige.write(bytes, 0, bytes.length);
}
}
/**
* 在close方法中關閉輸出流。
*/
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException {
IOUtils.closeQuietly(out_jige);
IOUtils.closeQuietly(out_bujige);
}
}