一：背景

有時候，我們不想再程式中顯示的指定輸入路徑和輸出路徑，因為那樣不太靈活，不利於擴充套件，Hadoop提供了將程式打成jar包發到叢集上通過命令列引數指定輸入輸出路徑的方式執行程式。

二：技術實現

(1)：主類繼承Configured類還要實現Tool介面。

(2)：將我們以前寫的設定各種引數的程式碼寫在run()方法中(實現介面必須要實現run方法)。

(3)：還有一句很關鍵的程式碼就是：job.setJarByClass(XXX.class);即以Jar包的形式執行。

我們以單詞計數為例，有以下兩種方法！！！

方法一：繼承Configured類和實現Tool介面

public class WordCount extends Configured implements Tool {

	// 定義輸入路徑
	private String INPUT_PATH = "";
	// 定義輸出路徑
	private String OUT_PATH = "";

	public static void main(String[] args) {

		try {
			ToolRunner.run(new WordCount(), args);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException,
				InterruptedException {

			// 注：value是hadoop的Text型別，呼叫toString可以轉換成java的型別。
			String[] splited = value.toString().split("\t");

			// 迭代
			for (String word : splited) {
				context.write(new Text(word), new LongWritable(1L));
			}
		}
	}

	static class MyReduce extends Reducer<Text, LongWritable, Text, LongWritable> {

		@Override
		protected void reduce(Text k2, Iterable<LongWritable> v2s, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException,
				InterruptedException {

			long sum = 0L;
			// 迭代計算單詞在檔案中出現的總記錄數
			for (LongWritable v2 : v2s) {
				sum += v2.get();
			}

			// 寫到上下文中
			context.write(k2, new LongWritable(sum));
		}
	}

	public int run(String[] args) throws Exception {

		// 給路徑賦值
		INPUT_PATH = args[0];
		OUT_PATH = args[1];

		try {
			// 建立配置資訊
			Configuration conf = new Configuration();
			// 新增配置檔案(我們可以在程式設計的時候動態配置資訊，而不需要手動去改變叢集)

			//conf.addResource("classpath://hadoop/core-site.xml");
			//conf.addResource("classpath://hadoop/hdfs-site.xml");
			//conf.addResource("classpath://hadoop/mapred-site.xml");

			// 如果輸出目錄存在，我們就進行刪除
			FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
			if (fileSystem.exists(new Path(OUT_PATH))) {
				fileSystem.delete(new Path(OUT_PATH), true);
			}

			// 建立任務
			Job job = new Job(conf, WordCount.class.getSimpleName());
			// 通過命令列傳參的形式必須走這一步(打包執行必須執行的關鍵)
			job.setJarByClass(WordCount.class);

			// 1.1 設定輸入目錄
			FileInputFormat.setInputPaths(job, INPUT_PATH);
			// 指定對輸入資料進行格式化處理的類(這個可以省略，預設值即可)
			job.setInputFormatClass(TextInputFormat.class);

			// 1.2 指定自定義的Mapper類
			job.setMapperClass(MyMapper.class);
			// 指定map輸出的<K,V>型別(如果<K3,V3>的型別和<K2,V2>的型別一致，那麼可以省略)
			// job.setMapOutputKeyClass(Text.class);
			// job.setMapOutputValueClass(LongWritable.class);

			// 1.3 分割槽(可以省略，預設即可)
			job.setPartitionerClass(HashPartitioner.class);
			job.setNumReduceTasks(1);// 預設的分割槽是1個，所以任務也就是1個

			// 1.4排序、分組

			// 1.5 歸約

			// 2.1 對多個map任務的輸出，按照不同的分割槽，通過網路copy到不同reduce節點。

			// 2.2 指定自定義的reduce類
			job.setReducerClass(MyReduce.class);
			// 指定<K,V>的型別
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(LongWritable.class);

			// 2.3 指定輸出的路徑
			FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
			// 指定輸出的格式化類(這個可以省略，使用預設值即可)
			job.setOutputFormatClass(TextOutputFormat.class);

			// 把作業提交給JobTracker完成
			job.waitForCompletion(true);

		} catch (Exception e) {
			e.printStackTrace();
		}

		return 0;
	}
}
 

上傳到遠端伺服器，通過如下命令執行：

我們可以先用jar -tf XXX.jar檢視jar中的結構，如下

然後用hadoop jar命令執行，如果我們是直接把整個專案打成一個jar包(因為很多時候我們會將類寫在多個檔案中，所以要打包整個專案),我們就要加上包名.主類名。如：

hadoop jar WordCount.jar com.lixue.run.cmd.WordCount  /hello  /out 

由於我們這個例子是直接將主類進行了打包，所以我們可以直接執行jar包就可以了，如下：

hadoop jar WordCount.jar hdfs://liaozhongmin5:9000/hello hdfs://liaozhongmin5:9000/out
 

方法二：也可以不繼承Configured和實現Tool介面，程式碼如下：

public class WordCountTest {
	
	//定義輸入路徑
	private static String IN_PATH = "";
	//定義輸出路徑
	private static String OUT_PATH = "";

	public static void main(String[] args) {

		try {
			// 建立配置資訊
			Configuration conf = new Configuration();
			//獲取命令列的引數
			String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
			//當引數違法時，中斷程式
			if (otherArgs.length != 2){
				System.err.println("Usage:wordcount<in> <out>");
				System.exit(1);
			}
			
			//給路徑賦值
			IN_PATH = otherArgs[0];
			OUT_PATH = otherArgs[1];
			// 建立檔案系統
			FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);
			// 如果輸出目錄存在，我們就刪除
			if (fileSystem.exists(new Path(new URI(OUT_PATH)))) {
				fileSystem.delete(new Path(new URI(OUT_PATH)), true);
			}

			// 建立任務
			Job job = new Job(conf, WordCountTest.class.getName());
			//打成jar包執行，這句話是關鍵
			job.setJarByClass(WordCountTest.class);
			//1.1	設定輸入目錄和設定輸入資料格式化的類
			FileInputFormat.setInputPaths(job, IN_PATH);
			job.setInputFormatClass(TextInputFormat.class);

			//1.2	設定自定義Mapper類和設定map函式輸出資料的key和value的型別
			job.setMapperClass(MyMapper.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(LongWritable.class);

			//1.3	設定分割槽和reduce數量(reduce的數量，和分割槽的數量對應，因為分割槽為一個，所以reduce的數量也是一個)
			job.setPartitionerClass(HashPartitioner.class);
			job.setNumReduceTasks(1);

			//1.4	排序
			//1.5	歸約
			//2.1	Shuffle把資料從Map端拷貝到Reduce端。
			//2.2	指定Reducer類和輸出key和value的型別
			job.setReducerClass(MyReducer.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(LongWritable.class);

			//2.3	指定輸出的路徑和設定輸出的格式化類
			FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
			job.setOutputFormatClass(TextOutputFormat.class);


			// 提交作業 退出
			System.exit(job.waitForCompletion(true) ? 0 : 1);
		
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

		// 定義一個LongWritable物件作為map輸出的value型別
		LongWritable oneTime = new LongWritable(1);
		// 定義一個Text物件作為map輸出的key型別
		Text word = new Text();

		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException,
				InterruptedException {

			// 對每一行記錄採用製表符(\t)進行分割
			String[] splits = value.toString().split("\t");

			// 遍歷字串陣列輸出每一個單詞
			for (String str : splits) {

				// 設定word
				word.set(str);
				// 把結果寫出去
				context.write(word, oneTime);
			}
		}
	}

	public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

		// 定義LongWritable物件最為Reduce輸出的value型別
		LongWritable result = new LongWritable();

		protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException,
				InterruptedException {

			int sum = 0;

			// 遍歷集合，計算每個單詞出現的和
			for (LongWritable s : values) {

				sum += s.get();
			}
			// 設定result
			result.set(sum);
			// 把結果寫出去
			context.write(key, result);
		}
	}
}
 

程式執行的日誌如下：

1.MapReduce任務的計算量即消耗CPU的時間=CPU time spent (ms)=1400

2.程式執行所消耗的實體記憶體=Physical memory (bytes) snapshot=184586240

3.程式執行所消耗的虛擬記憶體=Virtual memory (bytes) snapshot=756031488

4.JVM當前堆的大小=Total committed heap usage (bytes)=177016832

5.IO消耗=HDFS_BYTES_READ+HDFS_BYTES_WRITTEN*副本數+FILE_BYTES_READ+FILE_BYTES_WRITTEN=115+19*1+65+105096

6.網路流量消耗情況=HDFS_BYTES_READ+HDFS_BYTES_WRITTEN*副本數+Reduce shuffle bytes=115+19*1 + 65