Hadoop 程 式 設 計 五 開 發 Hadoop Map/Reduce 程 式 設 計 者 只 需 要 解 決 真 實 的 問 題, 架 構 面 留 給 MapReduce 96
Hadoop 的 MapReduce API 提 供 自 動 的 平 行 化 與 工 作 分 配 容 錯 特 性 狀 態 監 控 工 具 一 個 乾 淨 的 抽 象 化 (abstration) 供 程 式 設 計 師 使 用 97
HDFS & MapReduce HDFS Input Output 部 份 圖 片 來 源 :http://www.larsgeorge.com/2009/05/hbase-mapreduce-101-part-i.html 98
<Key, Value> Pair Input Row Data Map Output key1 key2 key1 val val val Map Select Key Reduce Input key1 val. val val Reduce Output key values 99
Program Prototype (v 0.20) Map 區 Reduce 區 設 定 區 Class MR{ static public Class Mapper { Map 程 式 碼 static public Class Reducer { Reduce 程 式 碼 main(){ Configuration conf = new Configuration(); Job job = new Job(conf, job name"); job.setjarbyclass(thismainclass.class); job.setmapperclass(mapper.class); job.setreduceclass(reducer.class); FileInputFormat.addInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); 其 他 的 設 定 參 數 程 式 碼 job.waitforcompletion(true); 100
Class Mapper (v 0.20) 1 2 3 4 5 6 7 8 9 import org.apache.hadoop.mapreduce.mapper; class MyMap extends Mapper < INPUT INPUT OUTPUT OUTPUT KEY, VALUE, KEY, VALUE > { Class Class Class Class // 全 域 變 數 區 INPUT INPUT public void map ( KEY key, VALUE value, Class Class Context context )throws IOException,InterruptedException { // 區 域 變 數 與 程 式 邏 輯 區 context.write( NewKey, NewValue); 101
Class Reducer (v 0.20) 1 2 3 4 5 6 7 8 9 import org.apache.hadoop.mapreduce.reducer; MyRed class MyRed extends INPUT INPUT OUTPUT OUTPUT Reducer <,,, > KEY VALUE KEY VALUE { Class Class Class Class // 全 域 變 數 區 public void reduce ( INPUT KEY Class key, Iterable< VALUE > values, Context context) throws IOException,InterruptedException { // 區 域 變 數 與 程 式 邏 輯 區 context.write( NewKey, NewValue); INPUT Class 102
其 他 常 用 的 設 定 參 數 設 定 Combiner Job.setCombinerClass ( ); 設 定 output class Job.setMapOutputKeyClass( ); Job.setMapOutputValueClass( ); Job.setOutputKeyClass( ); Job.setOutputValueClass( ); 103
Class Combiner 指 定 一 個 combiner, 它 負 責 對 中 間 過 程 的 輸 出 進 行 聚 集, 這 會 有 助 於 降 低 從 Mapper 到 Reducer 數 據 傳 輸 量 可 不 用 設 定 交 由 Hadoop 預 設 也 可 不 實 做 此 程 式, 引 用 Reducer 設 定 JobConf.setCombinerClass(Class) 104
範 例 一 (1) - mapper public class HelloHadoop { static public class HelloMapper extends Mapper<LongWritable, Text, LongWritable, Text> { public void map(longwritable key, Text value, Context context) throws IOException, InterruptedException { context.write((longwritable) key, (Text) value); // HelloReducer end..( 待 續 ) 105
範 例 一 (2) - reducer static public class HelloReducer extends Reducer<LongWritable, Text, LongWritable, Text> { public void reduce(longwritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException { Text val = new Text(); for (Text str : values) { val.set(str.tostring()); context.write(key, val); // HelloReducer end..( 待 續 ) 106
範 例 一 (3) - main public static void main(string[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); Job job = new Job(conf, "Hadoop Hello World"); job.setjarbyclass(hellohadoop.class); FileInputFormat.setInputPaths(job, "input"); FileOutputFormat.setOutputPath(job, new Path("output-hh1")); job.setmapperclass(hellomapper.class); job.setreducerclass(helloreducer.class); job.waitforcompletion(true); // main end // wordcount class end // 完 107
Hadoop 程 式 設 計 七 Hadoop 程 式 範 例 7.1:HDFS 操 作 篇 7.2:MapReduce 運 算 篇 108
傳 送 檔 案 至 HDFS // 將 檔 案 從 local 上 傳 到 hdfs, src 為 local 的 來 源, dst 為 hdfs 的 目 的 端 public class PutToHdfs { static boolean puttohdfs(string src, String dst, Configuration conf) { Path dstpath = new Path(dst); try { // 產 生 操 作 hdfs 的 物 件 FileSystem hdfs = dstpath.getfilesystem(conf); // 上 傳 hdfs.copyfromlocalfile(false, new Path(src),new Path(dst)); catch (IOException e) { e.printstacktrace(); return false; return true; 109
從 HDFS 取 回 檔 案 // 將 檔 案 從 hdfs 下 載 回 local, src 為 hdfs 的 來 源, dst 為 local 的 目 的 端 public class GetFromHdfs { static boolean getfromhdfs(string src,string dst, Configuration conf) { Path dstpath = new Path(src); try { // 產 生 操 作 hdfs 的 物 件 FileSystem hdfs = dstpath.getfilesystem(conf); // 下 載 hdfs.copytolocalfile(false, new Path(src),new Path(dst)); catch (IOException e) { e.printstacktrace(); return false; return true; 110
檢 查 與 刪 除 檔 案 // checkanddelete 函 式, 檢 查 是 否 存 在 該 資 料 夾, 若 有 則 刪 除 之 public class CheckAndDelete { static boolean checkanddelete(final String path, Configuration conf) { Path dst_path = new Path(path); try { // 產 生 操 作 hdfs 的 物 件 FileSystem hdfs = dst_path.getfilesystem(conf); // 檢 查 是 否 存 在 if (hdfs.exists(dst_path)) { // 有 則 刪 除 hdfs.delete(dst_path, true); catch (IOException e) { e.printstacktrace(); return false; return true; 111
Hadoop 程 式 設 計 七 Hadoop 程 式 範 例 7.1:HDFS 操 作 篇 7.2:MapReduce 運 算 篇 112
範 例 二 (1) HelloHadoopV2 說 明 : 此 程 式 碼 比 HelloHadoop 增 加 了 * 檢 查 輸 出 資 料 夾 是 否 存 在 並 刪 除 * input 資 料 夾 內 的 資 料 若 大 於 兩 個, 則 資 料 不 會 被 覆 蓋 * map 與 reduce 拆 開 以 利 程 式 再 利 用 測 試 方 法 : 將 此 程 式 運 作 在 hadoop 0.20 平 台 上, 執 行 : --------------------------- hadoop jar V2.jar HelloHadoopV2 --------------------------- 注 意 : 1. 在 hdfs 上 來 源 檔 案 的 路 徑 為 "/user/$your_name/input", 請 注 意 必 須 先 放 資 料 到 此 hdfs 上 的 資 料 夾 內, 且 此 資 料 夾 內 只 能 放 檔 案, 不 可 再 放 資 料 夾 2. 運 算 完 後, 程 式 將 執 行 結 果 放 在 hdfs 的 輸 出 路 徑 為 "/user/$your_name/output-hh2" 113
範 例 二 (2) public class HelloHadoopV2 { public static void main(string[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); Job job = new Job(conf, "Hadoop Hello World 2"); job.setjarbyclass(hellohadoopv2.class); // 設 定 map and reduce 以 及 Combiner class job.setmapperclass(hellomapperv2.class); job.setcombinerclass(helloreducerv2.class); job.setreducerclass(helloreducerv2.class); // 設 定 map 的 輸 出 型 態 job.setmapoutputkeyclass(text.class); job.setmapoutputvalueclass(text.class); // 設 定 reduce 的 輸 出 型 態 job.setoutputkeyclass(text.class); job.setoutputvalueclass(text.class); FileInputFormat.addInputPath (job, new Path("input")); FileOutputFormat.setOutputPath (job, new Path("output-hh2")); // 呼 叫 checkanddelete 函 式, // 檢 查 是 否 存 在 該 資 料 夾, 若 有 則 刪 除 之 CheckAndDelete.checkAndDelete("output-hh2", conf); boolean status = job.waitforcompletion(true); if (status) { System.err.println("Integrate Alert Job Finished!"); else { System.err.println("Integrate Alert Job Failed!"); System.exit(1); 114
範 例 二 (3) public class HelloMapperV2 extends Mapper <LongWritable, Text, Text, Text> { public void map(longwritable key, Text value, Context context) throws IOException, InterruptedException { context.write(new Text(key.toString()), value); public class HelloReducerV2 extends Reducer<Text, Text, Text, Text> { public void reduce(text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { String str = new String(""); Text final_key = new Text(); Text final_value = new Text(); // 將 key 值 相 同 的 values, 透 過 && 符 號 分 隔 之 for (Text tmp : values) { str += tmp.tostring() + " &&"; final_key.set(key); final_value.set(str); context.write(final_key, final_value); 115
範 例 三 (1) HelloHadoopV3 說 明 : 此 程 式 碼 再 利 用 了 HelloHadoopV2 的 map, reduce 檔, 並 且 自 動 將 檔 案 上 傳 到 hdfs 上 運 算 並 自 動 取 回 結 果, 還 有 提 示 訊 息 參 數 輸 入 與 印 出 運 算 時 間 的 功 能 測 試 方 法 : 將 此 程 式 運 作 在 hadoop 0.20 平 台 上, 執 行 : --------------------------- hadoop jar V3.jar HelloHadoopV3 <local_input> <local_output> --------------------------- 注 意 : 1. 第 一 個 輸 入 的 參 數 是 在 local 的 輸 入 資 料 夾, 請 確 認 此 資 料 夾 內 有 資 料 並 無 子 目 錄 2. 第 二 個 輸 入 的 參 數 是 在 local 的 運 算 結 果 資 料 夾, 由 程 式 產 生 不 用 事 先 建 立, 若 有 請 刪 除 之 116
範 例 三 (2) public class HelloHadoopV3 { public static void main(string[] args) throws IOException, InterruptedException, ClassNotFoundException { 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 <local_input> <local_output>"); System.exit(2); Job job = new Job(conf, "Hadoop Hello World"); job.setjarbyclass(hellohadoopv3.class); // 再 利 用 上 個 範 例 的 map 與 reduce job.setmapperclass(hellomapperv2.class); job.setcombinerclass(helloreducerv2.class); job.setreducerclass(helloreducerv2.class); // 設 定 map reduce 的 key value 輸 出 型 態 job.setmapoutputkeyclass(text.class); job.setmapoutputvalueclass(text.class); job.setoutputkeyclass(text.class); job.setoutputvalueclass(text.class); 117
範 例 三 (2) // 用 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); 118
範 例 四 (1) public static void main(string[] args) throws Exception { Configuration conf = new Configuration(); String[] otherargs = new GenericOptionsParser(conf, args).getremainingargs(); if (otherargs.length!= 2) { System.err.println("Usage: hadoop jar WordCount.jar <input> <output>"); System.exit(2); Job job = new Job(conf, "Word Count"); job.setjarbyclass(wordcount.class); job.setmapperclass(tokenizermapper.class); job.setcombinerclass(intsumreducer.class); job.setreducerclass(intsumreducer.class); job.setoutputkeyclass(text.class); job.setoutputvalueclass(intwritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); CheckAndDelete.checkAndDelete(args[1], conf); System.exit(job.waitForCompletion(true)? 0 : 1); 119
範 例 四 (2) 1 class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 2 private final static IntWritable one = new IntWritable(1); 3 private Text word = new Text(); 4 public void map( LongWritable key, Text value, Context context) 5 6 7 8 9 Input key /user/hadooper/input/a.txt. No news is a good news. throws IOException, InterruptedException { String line = ((Text) value).tostring(); StringTokenizer itr = new StringTokenizer(line); while (itr.hasmoretokens()) { word.set(itr.nexttoken()); context.write(word, one); Input value line itr no news is a good news itr itr itr itr itr itr <word,one> < no, 1 > < news, 1 > < is, 1 > < a, 1 > < good, 1 > < news, 1 > 120
範 例 四 (3) 1 2 3 4 5 6 7 8 class IntSumReducer extends Reducer< Text, IntWritable, Text, IntWritable> { 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); for ( int i ; i < values.length ; i ++ ){ sum += values[i].get() <word,one> < a, 1 > < good, 1 > < is, 1 > < news, 1 1 > < no, 1 > news 1 1 <key,sunvalue> < news, 2 > 121
範 例 五 (1) WordCountV2 說 明 : 用 於 字 數 統 計, 並 且 增 加 略 過 大 小 寫 辨 識 符 號 篩 除 等 功 能 測 試 方 法 : 將 此 程 式 運 作 在 hadoop 0.20 平 台 上, 執 行 : --------------------------- hadoop jar WCV2.jar WordCountV2 -Dwordcount.case.sensitive=false \ <input> <output> -skip patterns/patterns.txt --------------------------- 注 意 : 1. 在 hdfs 上 來 源 檔 案 的 路 徑 為 你 所 指 定 的 <input> 請 注 意 必 須 先 放 資 料 到 此 hdfs 上 的 資 料 夾 內, 且 此 資 料 夾 內 只 能 放 檔 案, 不 可 再 放 資 料 夾 2. 運 算 完 後, 程 式 將 執 行 結 果 放 在 hdfs 的 輸 出 路 徑 為 你 所 指 定 的 <output> 3. 請 建 立 一 個 資 料 夾 pattern 並 在 裡 面 放 置 pattern.txt, 內 容 如 ( 一 行 一 個, 前 置 提 示 符 號 \) \. \, \! 122
範 例 五 (2) public class WordCountV2 extends Configured implements Tool { public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> { static enum Counters { INPUT_WORDS private final static IntWritable one = new IntWritable(1); private Text word = new Text(); private boolean casesensitive = true; private Set<String> patternstoskip = new HashSet<String>(); private long numrecords = 0; private String inputfile; public void configure(jobconf job) { casesensitive = job.getboolean("wordcount.case.sensitive", true); inputfile = job.get("map.input.file"); if (job.getboolean("wordcount.skip.patterns", false)) { Path[] patternsfiles = new Path[0]; try { patternsfiles = DistributedCache.getLocalCacheFiles(job); catch (IOException ioe) { System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe)); for (Path patternsfile : patternsfiles) { parseskipfile(patternsfile); private void parseskipfile(path patternsfile) { try { BufferedReader fis = new BufferedReader(new FileReader( patternsfile.tostring())); String pattern = null; while ((pattern = fis.readline())!= null) { patternstoskip.add(pattern); catch (IOException ioe) { System.err.println("Caught exception while parsing the cached file '"+ patternsfile + "' : " + tringutils.stringifyexception(ioe)); public void map(longwritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { String line = (casesensitive)? value.tostring() : value.tostring().tolowercase(); for (String pattern : patternstoskip) line = line.replaceall(pattern, ""); StringTokenizer tokenizer = new StringTokenizer(line); while (tokenizer.hasmoretokens()) { word.set(tokenizer.nexttoken()); output.collect(word, one); reporter.incrcounter(counters.input_words, 1); 123
範 例 五 (3) if ((++numrecords % 100) == 0) { reporter.setstatus("finished processing " + numrecords + " records " + "from the input file: " + inputfile); public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> { public void reduce(text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException { int sum = 0; while (values.hasnext()) { sum += values.next().get(); output.collect(key, new IntWritable(sum)); public int run(string[] args) throws Exception { JobConf conf = new JobConf(getConf(), WordCount.class); conf.setjobname("wordcount"); String[] otherargs = new GenericOptionsParser(conf, args).getremainingargs(); if (otherargs.length < 2) { System.out.println("WordCountV2 [- Dwordcount.case.sensitive=<false true>] \\ "); System.out.println(" <indir> <outdir> [-skip Pattern_file]"); return 0; conf.setoutputkeyclass(text.class); conf.setoutputvalueclass(intwritable.class); conf.setmapperclass(map.class); conf.setcombinerclass(reduce.class); conf.setreducerclass(reduce.class); conf.setinputformat(textinputformat.class); conf.setoutputformat(textoutputformat.class); List<String> other_args = new ArrayList<String>(); for (int i = 0; i < args.length; ++i) { if ("-skip".equals(args[i])) { DistributedCache.addCacheFile(new Path(args[++i]).toUri(), conf); conf.setboolean("wordcount.skip.patterns", true); else {other_args.add(args[i]); FileInputFormat.setInputPaths(conf, new Path(other_args.get(0))); FileOutputFormat.setOutputPath(conf, new Path(other_args.get(1))); CheckAndDelete.checkAndDelete(other_args.get(1), conf); JobClient.runJob(conf); return 0; public static void main(string[] args) throws Exception { int res = ToolRunner.run(new Configuration(), new WordCountV2(), args); System.exit(res); 124
說 明 : 範 例 六 (1) WordIndex 將 每 個 字 出 於 哪 個 檔 案, 那 一 行 印 出 來 測 試 方 法 : 將 此 程 式 運 作 在 hadoop 0.20 平 台 上, 執 行 : --------------------------- hadoop jar WI.jar WordIndex <input> <output> --------------------------- 注 意 : 1. 在 hdfs 上 來 源 檔 案 的 路 徑 為 你 所 指 定 的 <input> 請 注 意 必 須 先 放 資 料 到 此 hdfs 上 的 資 料 夾 內, 且 此 資 料 夾 內 只 能 放 檔 案, 不 可 再 放 資 料 夾 2. 運 算 完 後, 程 式 將 執 行 結 果 放 在 hdfs 的 輸 出 路 徑 為 你 所 指 定 的 <output> 125
public class WordIndex { public static class wordindexm extends Mapper<LongWritable, Text, Text, Text> { public void map(longwritable key, Text value, Context context) throws IOException, InterruptedException { FileSplit filesplit = (FileSplit) context.getinputsplit(); 範 例 六 (2) static public class wordindexr extends Reducer<Text, Text, Text, Text> { public void reduce(text key, Iterable<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException { String v = ""; Text map_key = new Text(); StringBuilder ret = new StringBuilder("\n"); Text map_value = new Text(); for (Text val : values) { String line = value.tostring(); v += val.tostring().trim(); StringTokenizer st = new StringTokenizer(line.toLowerCase()); if (v.length() > 0) while (st.hasmoretokens()) { ret.append(v + "\n"); String word = st.nexttoken(); map_key.set(word); output.collect((text) key, new map_value.set(filesplit.getpath().getname() + Text(ret.toString())); ":" + line); context.write(map_key, map_value); 126
範 例 六 (2) public static void main(string[] args) throws IOException, InterruptedException, ClassNotFoundException { Configuration conf = new Configuration(); String[] otherargs = new GenericOptionsParser(conf, args).getremainingargs(); if (otherargs.length < 2) { System.out.println("hadoop jar WordIndex.jar <indir> <outdir>"); return; Job job = new Job(conf, "word index"); job.setjobname("word inverted index"); job.setjarbyclass(wordindex.class); job.setmapoutputkeyclass(text.class); job.setmapoutputvalueclass(text.class); job.setoutputkeyclass(text.class); job.setoutputvalueclass(text.class); job.setmapperclass(wordindexm.class); job.setreducerclass(wordindexr.class); job.setcombinerclass(wordindexr.class); FileInputFormat.setInputPaths(job, args[0]); CheckAndDelete.checkAndDelete(args[1], conf); FileOutputFormat.setOutputPath(job, new Path(args[1])); long start = System.nanoTime(); job.waitforcompletion(true); long time = System.nanoTime() - start; System.err.println(time * (1E-9) + " secs."); 127
範 例 七 (1) YourMenu 說 明 : 將 之 前 的 功 能 整 合 起 來 測 試 方 法 : 將 此 程 式 運 作 在 hadoop 0.20 平 台 上, 執 行 : --------------------------- hadoop jar YourMenu.jar < 功 能 > --------------------------- 注 意 : 1. 此 程 式 需 與 之 前 的 所 有 範 例 一 起 打 包 成 一 個 jar 檔 128
範 例 七 (2) public class YourMenu { public static void main(string argv[]) { int exitcode = -1; ProgramDriver pgd = new ProgramDriver(); if (argv.length < 1) { System.out.print("********************************** ********\n" + " 歡 迎 使 用 NCHC 的 運 算 功 能 \n" + " 指 令 : \n" + " Hadoop jar NCHC-example-*.jar < 功 能 > \n" + " 功 能 : \n" + " HelloHadoop: 秀 出 Hadoop 的 <Key,Value> 為 何 \n" + " HelloHadoopV2: 秀 出 Hadoop 的 <Key,Value> 進 階 版 \n" + " HelloHadoopV3: 秀 出 Hadoop 的 <Key,Value> 進 化 版 \n" + " WordCount: 計 算 輸 入 資 料 夾 內 分 別 在 每 個 檔 案 的 字 數 統 計 \n" + " WordCountV2: WordCount 進 階 版 \n" + " WordIndex: 索 引 每 個 字 與 其 所 有 出 現 的 所 在 列 \n" + "******************************************\n"); else { try { pgd.addclass("hellohadoop", HelloHadoop.class, " Hadoop hello world"); pgd.addclass("hellohadoopv2", HelloHadoopV2.class, " Hadoop hello world V2"); pgd.addclass("hellohadoopv3", HelloHadoopV3.class, " Hadoop hello world V3"); pgd.addclass("wordcount", WordCount.class, " word count."); pgd.addclass("wordcountv2", WordCountV2.class, " word count V2."); pgd.addclass("wordindex", WordIndex.class, "invert each word in line"); pgd.driver(argv); // Success exitcode = 0; System.exit(exitCode); catch (Throwable e) { e.printstacktrace(); 129
補 充 Program Prototype (v 0.18) Map 區 Reduce 區 設 定 區 Class MR{ Class Mapper { Class Reducer { main(){ Map 程 式 碼 Reduce 程 式 碼 JobConf conf = new JobConf( MR.class ); conf.setmapperclass(mapper.class); conf.setreduceclass(reducer.class); FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); 其 他 的 設 定 參 數 程 式 碼 JobClient.runJob(conf); 130
補 充 Class Mapper (v0.18) 1 2 3 4 5 6 7 8 9 import org.apache.hadoop.mapred.*; MyMap class MyMap extends MapReduceBase INPUT INPUT OUTPUT OUTPUT implements Mapper < KEY, VALUE, KEY, VALUE > { // 全 域 變 數 區 INPUT INPUT public void map ( KEY key, VALUE value, OUTPUT OUTPUT OutputCollector<, > output, KEY VALUE Reporter reporter) throws IOException { // 區 域 變 數 與 程 式 邏 輯 區 output.collect( NewKey, NewValue); 131
補 充 Class Reducer (v0.18) import org.apache.hadoop.mapred.*; 1 2 3 4 5 6 7 8 9 class MyRed extends MapReduceBase INPUT INPUT OUTPUT OUTPUT implements Reducer < KEY, VALUE, KEY, VALUE > { // 全 域 變 數 區 INPUT KEY INPUT public void reduce ( key, Iterator< VALUE > values, OUTPUT OUTPUT OutputCollector<, > output, KEY VALUE Reporter reporter) throws IOException { // 區 域 變 數 與 程 式 邏 輯 區 output.collect( NewKey, NewValue); 132
Conclusions 以 上 範 例 程 式 碼 包 含 Hadoop 的 key,value 架 構 操 作 Hdfs 檔 案 系 統 Map Reduce 運 算 方 式 執 行 hadoop 運 算 時, 程 式 檔 不 用 上 傳 至 hadoop 上, 但 資 料 需 要 再 HDFS 內 可 運 用 範 例 七 的 程 式 達 成 連 續 運 算 Hadoop 0.20 與 Hadoop 0.18 有 些 API 有 些 許 差 異, 因 此 在 網 路 上 找 到 Hadoop 的 程 式 如 果 compiler 有 錯, 可 以 換 換 對 應 的 Function 試 試 133