{{{ #!html
Hadoop 進階課程
練習
}}} [wiki:NCHCCloudCourse100928_4_EXM2 上一關 < ] 第三關 [wiki:NCHCCloudCourse100928_4_EXM4 > 下一關] = 說明 = {{{ #!text HelloHadoopV3 說明: 此程式碼再利用了 HelloHadoopV2 的 map , reduce 檔,並且 自動將檔案上傳到hdfs上運算並自動取回結果,還有 提示訊息 、參數輸入 與 印出運算時間 的功能 測試方法: 將此程式運作在hadoop 0.20 平台上,執行: --------------------------- hadoop jar HelloHadoopV3.jar /home/$yourname/input /home/$yourname/output-hh3 --------------------------- 注意: 1. 第一個輸入的參數是在local 的 輸入資料夾,請確認此資料夾內有資料並無子目錄 2. 第二個輸入的參數是在local 的 運算結果資料夾,由程式產生不用事先建立,若有請刪除之 }}} = HelloHadoopV3.java = {{{ #!java package org.nchc.hadoop; import java.io.IOException; import org.apache.hadoop.conf.Configuration; 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.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class HelloHadoopV3 { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { // eclipse using // String[] argv = {"/home/hadoop/input","/home/hadoop/output-hh3"}; // args = argv; String hdfs_input = "HH3_input"; String hdfs_output = "HH3_output"; Configuration conf = new Configuration(); // 宣告取得參數 String[] otherArgs = new GenericOptionsParser(conf, args) .getRemainingArgs(); // 如果參數數量不為2 則印出提示訊息 if (otherArgs.length != 2) { System.err .println("Usage: hadoop jar HelloHadoopV3.jar "); System.exit(2); } Job job = new Job(conf, "Hadoop Hello World"); job.setJarByClass(HelloHadoopV3.class); // set map and reduce class job.setMapperClass(); // 輸入正確的mapper job.setCombinerClass(); // 輸入正確的 class job.setReducerClass(); // 輸入正確的 reducer job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); // 用 checkAndDelete 函式防止overhead的錯誤 CheckAndDelete.checkAndDelete(hdfs_input, conf); CheckAndDelete.checkAndDelete(hdfs_output, conf); // 放檔案到hdfs PutToHdfs.putToHdfs(args[0], hdfs_input, conf); // 設定hdfs 的輸入輸出來源路定 FileInputFormat.addInputPath(job, new Path(hdfs_input)); FileOutputFormat.setOutputPath(job, new Path(hdfs_output)); long start = System.nanoTime(); job.waitForCompletion(true); // 把hdfs的結果取下 GetFromHdfs.getFromHdfs(hdfs_output, args[1], conf); boolean status = job.waitForCompletion(true); // 計算時間 if (status) { System.err.println("Integrate Alert Job Finished !"); long time = System.nanoTime() - start; System.err.println(time * (1E-9) + " secs."); } else { System.err.println("Integrate Alert Job Failed !"); System.exit(1); } } } }}}