package cn.yzx.bigdata.mr.wcemo;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


/*
 * 為啥要main方法：先啟動自己，然後把整個程式提交到叢集
 * 相當於yarn叢集的客戶端，yarn去分配運算資源，然後才能啟動
 * 以為是yarn的客戶端，所以要再次封裝mr程式的相關執行引數，指定jar包
 * 最後提交給yarn
 */
public class WordcountDriver {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		/*
		 * import org.apache.hadoop.mapreduce.xxxx
		 * import org.apache.hadoop.mapred.xxx  mapred是老版本的，我們用MapReduce 
		 */
		Configuration conf = new Configuration();
		/*conf.set("mapreduce.framework.name", "yarn");
		conf.set("yarn.resourcemanager.hostname","hadoop01");*/
		//給一些預設的引數
		Job job = Job.getInstance(conf);
		
		
		
		
		//指定本程式的jar包所在的本地路徑  把jar包提交到yarn
		job.setJarByClass(WordcountDriver.class);
		
		
		
		
		/*
		 * 告訴框架呼叫哪個類
		 * 指定本業務job要是用的mapper/Reducer業務類
		 */
		job.setMapperClass(WordcountMapper.class);
		job.setReducerClass(WordcountReduce.class);


		
		
		/*
		 * 指定mapper輸出資料KV型別
		 */
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//指定最終的輸出資料的kv型別  ，有時候不需要reduce過程，如果有的話最終輸出指的就是指reducekv型別  
		job.setOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		
		
		
		// 指定job的檔案輸入的原始目錄 
		//paths指你的待處理檔案可以在多個目錄裡邊
		//第一個引數是你給那個job設定  後邊的引數 逗號分隔的多個路徑 路徑是在hdfs裡的
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		
		// 指定job 的輸出結果所在的目錄
		FileOutputFormat.setOutputPath(job, new Path(args[1]));	
		
		
		/*
		 * 找yarn通訊
		 * 將job中配置的引數，  以及job所用的java類所在的jar包提交給yarn去執行
		 */
		/*job.submit();*/
		// 引數表示程式執行完，告訴我們是否執行成功
		boolean res = job.waitForCompletion(true);
		System.exit(res?0:1);
		
	}
}
 

1.2 WordcountMapper