Hadoop 的 MapReduce API 提供自動的平行化與工作分配容錯特性狀態監控工具一個乾淨的抽象化 (abstration) 供程式設計師使用 97

Size: px

Start display at page:

Download "Hadoop 的 MapReduce API 提供自動的平行化與工作分配容錯特性狀態監控工具一個乾淨的抽象化 (abstration) 供程式設計師使用 97"

崎俗甘
7 years ago
Views:

1 Hadoop 程式設計五開發 Hadoop Map/Reduce 程式設計者只需要解決真實的問題, 架構面留給 MapReduce 96

2 Hadoop 的 MapReduce API 提供自動的平行化與工作分配容錯特性狀態監控工具一個乾淨的抽象化 (abstration) 供程式設計師使用 97

3 HDFS & MapReduce HDFS Input Output 部份圖片來源 : 98

4 <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

5 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

Configuration conf = new Configuration(); Job job = new Job(conf, job name"); job.setjarbyclass(thismainclass.class); job.

6 Class Mapper (v 0.20) 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 Class Class // 全域變數區 INPUT INPUT public void map ( KEY key, VALUE value, Class Class

7 Class Reducer (v 0.20) 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

Class Class Class Class // 全域變數區 public void reduce ( INPUT KEY Class key, Iterable< VALUE >

8 其他常用的設定參數設定 Combiner Job.setCombinerClass ( ); 設定 output class Job.setMapOutputKeyClass( ); Job.setMapOutputValueClass( ); Job.setOutputKeyClass( ); Job.setOutputValueClass( ); 103

9 Class Combiner 指定一個 combiner, 它負責對中間過程的輸出進行聚集, 這會有助於降低從 Mapper 到 Reducer 數據傳輸量可不用設定交由 Hadoop 預設也可不實做此程式, 引用 Reducer 設定 JobConf.setCombinerClass(Class) 104

10 範例一 (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

map(longwritable key, Text value, Context context) throws IOException,

11 範例一 (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

context) throws IOException, InterruptedException { Text val = new Text(); for (Text str :

12 範例一 (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

class); FileInputFormat.setInputPaths(job, "input"); FileOutputFormat.setOutputPath(job, new Path("output-hh1")); job.

13 Hadoop 程式設計七 Hadoop 程式範例 7.1:HDFS 操作篇 7.2:MapReduce 運算篇 108

14 傳送檔案至 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 的物件 FileSystem hdfs = dstpath.getfilesystem(conf); // 上傳 hdfs.

15 從 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

try { // 產生操作 hdfs 的物件 FileSystem hdfs = dstpath.getfilesystem(conf); // 下載 hdfs.

16 檢查與刪除檔案 // 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

$exists(dst_path)) { // 有則刪除 hdfs.delete(dst_path, true); catch (IOException e) { e.$

17 Hadoop 程式設計七 Hadoop 程式範例 7.1:HDFS 操作篇 7.2:MapReduce 運算篇 112

18 範例二 (1) HelloHadoopV2 說明 : 此程式碼比 HelloHadoop 增加了 * 檢查輸出資料夾是否存在並刪除 * input 資料夾內的資料若大於兩個, 則資料不會被覆蓋 * map 與 reduce 拆開以利程式再利用測試方法 : 將此程式運作在 hadoop 0.20 平台上, 執行 : hadoop jar V2.jar HelloHadoopV 注意 : 1. 在 hdfs 上來源檔案的路徑為 "/user/$your_name/input", 請注意必須先放資料到此 hdfs 上的資料夾內, 且此資料夾內只能放檔案, 不可再放資料夾 2. 運算完後, 程式將執行結果放在 hdfs 的輸出路徑為 "/user/$your_name/output-hh2" 113

jar HelloHadoopV2 --------------------------- 注意 : 1.

19 範例二 (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

class); job.setreducerclass(helloreducerv2.class); // 設定 map 的輸出型態 job.setmapoutputkeyclass(text.class); job.setmapoutputvalueclass(text.class); // 設定 reduce 的輸出型態 job.

20 範例二 (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

$IOException, InterruptedException { String str = new String(""); Text final_key = new Text(); Text final_value = new Text(); // 將 key 值相同的 values, 透過 && 符號分隔之 for$

21 範例三 (1) HelloHadoopV3 說明 : 此程式碼再利用了 HelloHadoopV2 的 map, reduce 檔, 並且自動將檔案上傳到 hdfs 上運算並自動取回結果, 還有提示訊息參數輸入與印出運算時間的功能測試方法 : 將此程式運作在 hadoop 0.20 平台上, 執行 : hadoop jar V3.jar HelloHadoopV3 <local_input> <local_output> 注意 : 1. 第一個輸入的參數是在 local 的輸入資料夾, 請確認此資料夾內有資料並無子目錄 2. 第二個輸入的參數是在 local 的運算結果資料夾, 由程式產生不用事先建立, 若有請刪除之 116

22 範例三 (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

23 範例三 (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

24 範例四 (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

25 範例四 (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) 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

26 範例四 (3) 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

27 範例五 (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

28 範例五 (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

29 範例五 (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

30 說明 : 範例六 (1) WordIndex 將每個字出於哪個檔案, 那一行印出來測試方法 : 將此程式運作在 hadoop 0.20 平台上, 執行 : hadoop jar WI.jar WordIndex <input> <output> 注意 : 1. 在 hdfs 上來源檔案的路徑為你所指定的 <input> 請注意必須先放資料到此 hdfs 上的資料夾內, 且此資料夾內只能放檔案, 不可再放資料夾 2. 運算完後, 程式將執行結果放在 hdfs 的輸出路徑為你所指定的 <output> 125

31 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

32 範例六 (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

33 範例七 (1) YourMenu 說明 : 將之前的功能整合起來測試方法 : 將此程式運作在 hadoop 0.20 平台上, 執行 : hadoop jar YourMenu.jar < 功能 > 注意 : 1. 此程式需與之前的所有範例一起打包成一個 jar 檔 128

34 範例七 (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

35 補充 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

36 補充 Class Mapper (v0.18) 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

37 補充 Class Reducer (v0.18) import org.apache.hadoop.mapred.*; 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

38 Conclusions 以上範例程式碼包含 Hadoop 的 key,value 架構操作 Hdfs 檔案系統 Map Reduce 運算方式執行 hadoop 運算時, 程式檔不用上傳至 hadoop 上, 但資料需要再 HDFS 內可運用範例七的程式達成連續運算 Hadoop 0.20 與 Hadoop 0.18 有些 API 有些許差異, 因此在網路上找到 Hadoop 的程式如果 compiler 有錯, 可以換換對應的 Function 試試 133

Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采用分而治之的思想, 把对大规模数据集的操作, 分发给一个主节点管

Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采用分而治之的思想, 把对大规模数据集的操作, 分发给一个主节点管细细品味 Hadoop Hadoop 集群 ( 第 6 期 ) 精华集锦 csaxp 虾皮工作室 http://www.cnblogs.com/xia520pi/ 2012 年 5 月 15 日 Hadoop 集群 ( 第 6 期 ) WordCount 运行详解 1 MapReduce 理论简介 1.1 MapReduce 编程模型 MapReduce 采

Hadoop 的 MapReduce API 提 供 自 動 的 平 行 化 與 工 作 分 配 容 錯 特 性 狀 態 監 控 工 具 一 個 乾 淨 的 抽 象 化 (abstration) 供 程 式 設 計 師 使 用 97

Hadoop 的 MapReduce API 提供自動的平行化與工作分配容錯特性狀態監控工具一個乾淨的抽象化 (abstration) 供程式設計師使用 97