Hadoop MapReduce  Job 相關引數設定 概念介紹與理解

InputFormat:

作用：將輸入的檔案分成 一個個split,並且將split 拆分成一個個<key,value> 作為 map的輸入。

使用：通過job.setInputFormatClass() 進行設定

原理：

InputFormat 僅有兩個抽象方法： 

1、List<InputSplit> getSplits()  獲取檔案計算出的輸入分片，解決資料分割成片的問題。

2、RecordReader<K,V>createRecordReader()  從InputSplit 中讀取，解決從分片中讀取資料地方問題。將讀入Map的資料拆分成<key,value>類

public abstract class InputFormat<K, V> {
 
    /**
     * 僅僅是邏輯分片，並沒有物理分片，所以每一個分片類似於這樣一個元組 <input-file-path, start, offset>
     */
    public abstract List<InputSplit> getSplits(JobContext context)
            throws IOException, InterruptedException;
 
    /**
     * Create a record reader for a given split.
     */
    public abstract RecordReader<K, V> createRecordReader(InputSplit split,
            TaskAttemptContext context) throws IOException,
            InterruptedException;
 
}
 

import org.apache.hadoop.io.ArrayWritable;
import org.apache.hadoop.io.Text;

/**
 * 類說明
 * 
 * <pre>
 * Modify Information:
 * Author        Date          Description
 * ============ =========== ============================
 * DELL          2017年3月15日    Create this file
 * </pre>
 * 
 */

public class TextArrayWritable extends ArrayWritable {
    public TextArrayWritable() {
        super(Text.class);
    }

    public TextArrayWritable(Text[] strings) {
        super(Text.class, strings);
    }
}
 

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

/**
 * 類說明
 * 
 * <pre>
 * Modify Information:
 * Author        Date          Description
 * ============ =========== ============================
 * DELL          2017年3月15日    Create this file
 * </pre>
 * 
 */

public class CSVInputFormat extends FileInputFormat<LongWritable, TextArrayWritable> {
    public static final String CSV_TOKEN_SEPARATOR_CONFIG = "csvinputformat.token.delimiter";

    @Override
    protected boolean isSplitable(JobContext context, Path filename) {
        CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration()).getCodec(filename);
        return codec == null;
    }

    @Override
    public RecordReader<LongWritable, TextArrayWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException,
            InterruptedException {
        String csvDelimiter = context.getConfiguration().get(CSV_TOKEN_SEPARATOR_CONFIG);
        Character separator = null;
        if (csvDelimiter != null && csvDelimiter.length() == 1) {
            separator = csvDelimiter.charAt(0);
        }
        return new CSVRecordReader(separator);
    }
 
}

Mapper :

作用：實現map函式，根據輸入<key,value> 對生成中間結果。

使用：可以通過job.setMapperClass() 進行設定

原始碼：

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
 
    public class Context extends MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
        public Context(Configuration conf, TaskAttemptID taskid,
                RecordReader<KEYIN, VALUEIN> reader,
                RecordWriter<KEYOUT, VALUEOUT> writer,
                OutputCommitter committer, StatusReporter reporter,
                InputSplit split) throws IOException, InterruptedException {
            super(conf, taskid, reader, writer, committer, reporter, split);
        }
    }
 
    /**
     * 預處理，僅在map task啟動時執行一次
     */
    protected void setup(Context context) throws IOException,
            InterruptedException {
    }
 
    /**
     * 對於InputSplit中的每一對<key, value>都會執行一次
     */
    @SuppressWarnings("unchecked")
    protected void map(KEYIN key, VALUEIN value, Context context)
            throws IOException, InterruptedException {
        context.write((KEYOUT) key, (VALUEOUT) value);
    }
 
    /**
     * 掃尾工作，比如關閉流等
     */
    protected void cleanup(Context context) throws IOException,
            InterruptedException {
    }
 
    /**
     * map task的驅動器
     */
    public void run(Context context) throws IOException, InterruptedException {
        setup(context);
        while (context.nextKeyValue()) {
            map(context.getCurrentKey(), context.getCurrentValue(), context);
        }
        cleanup(context);
    }
}

例項：

 public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {

        private final static IntWritable one = new IntWritable(1);

        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            System.out.println("key: "+key.toString());
            System.out.println("value:"+value.toString());
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

作用：實現reduce 類，將中間結果合併，得到最終結果。

使用：job.setReducerClass()方法進行設定

使用例項：

  public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

Combiner :

作用： 實現map函式 合併中間結果中有相同key的鍵值對。

使用： job.setCombinerClass()方法進行設定， 預設為null 不合並， 輸出是reduce的輸入

例項使用：

    public static class Combine extends Reducer<Text,Text,Text,Text> {  
          
        // Reduce Method  
        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {  
            double sum = 0;  
            int count = 0;  
            for (Text value : values) {  
                String fields[] = value.toString().split(",");  
                sum += Double.parseDouble(fields[0]);  
                count += Integer.parseInt(fields[1]);  
            }  
            context.write(key, new Text(sum+","+count));  
        }  
    }  

partioner類：

作用： 實現getPartition()函式用來在Shuffle 過程中，按照key 將中間資料分成R份， 每一份由一個Reducer 負責。

使用：job.setPartitioner() 進行設定，預設是HashPartitoner

Shuffle 作用： 將Map 階段的輸出，Copy Reduce節點本地

使用例項：

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

/**
 * @author xadil
 *
 */
public class StatementsPartioner extends Partitioner<StatementsKeyType, Text> {

	@Override
	public int getPartition(StatementsKeyType key, Text value, int numPartitions) {
		
		return Math.abs(key.getAccountId().hashCode()*127)%numPartitions;
	}

}

作用： 負責輸出最終結果

使用：job.setOutputFormatClass() 進行設定， 預設是TextOutputFormat 類，TextOutputFormat 將結果寫成檔案。每行一個key -value 之間以製表符分開。

使用例項：

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author xadil
 *
 */
public class StatementPDFOutputFormat<K,V> extends FileOutputFormat<K,V> {
	public static String SEPERATOR = "mapreduce.output.textoutputformat.separator";

	/* (non-Javadoc)
	 * @see org.apache.hadoop.mapreduce.lib.output.FileOutputFormat#getRecordWriter(org.apache.hadoop.mapreduce.TaskAttemptContext)
	 */
	@Override
	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)
			throws IOException, InterruptedException {
		Configuration conf = job.getConfiguration();
		Path file = getDefaultWorkFile(job, "");
		
		return new PDFStatementWriter<K,V>(file,job.getConfiguration());
	}
	
	/*@Override
	public RecordWriter<K, V> getRecordWriter(
			TaskAttemptContext job) throws IOException, InterruptedException {
		Configuration conf = job.getConfiguration();
	    boolean isCompressed = getCompressOutput(job);
	    String keyValueSeparator= conf.get(SEPERATOR, "\t");
	    CompressionCodec codec = null;
	    String extension = "";
	    if (isCompressed) {
	      Class<? extends CompressionCodec> codecClass = 
	        getOutputCompressorClass(job, GzipCodec.class);
	      codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
	      extension = codec.getDefaultExtension();
	    }
	    Path file = getDefaultWorkFile(job, extension);
	    Path fullPath = new Path(file, "result.txt");
	    FileSystem fs = file.getFileSystem(conf);
	    if (!isCompressed) {
	      FSDataOutputStream fileOut = fs.create(fullPath, false);
	      return new PDFWriter<K, V>(fileOut, keyValueSeparator);
	    } else {
	      FSDataOutputStream fileOut = fs.create(fullPath, false);
	      return new PDFWriter<K, V>(new DataOutputStream
	                                        (codec.createOutputStream(fileOut)),
	                                        keyValueSeparator);
	    }
	}*/
	
	
	
}

作用：設定最終結果key的型別

OutputValue:

作用：設定最終結果value的型別。

MapOutputKey:

作用：設定中間結果（Mapper）key的型別

使用：方法類似

MapOutputValue:

作用：設定中間結果（Mapper）value的型別。

GroupingComparator:

作用： 設定分組函式類，相同key放到一起，value 放到同一個函式類。