CHAPTER 9 使用Hadoop實作MapReduce.

Presentation on theme: "CHAPTER 9 使用Hadoop實作MapReduce."— Presentation transcript:

1 CHAPTER 9 使用Hadoop實作MapReduce

2 Outline 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例
Hadoop MapReduce專題

3 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

4 開發環境設定 Hadoop程式開發環境的架設可分為兩種；
不透過Integrated Development Environment (IDE) 透過IDE

5 不使用IDE環境(1/2) 首先設定環境變數如下： 在/etc/profile中加入Java及Hadoop的CLASSPATH：
~# vi /etc/profile CLASSPATH=/opt/hadoop/hadoop core.jar ←加入這兩行 export CLASSPATH 透過source讓設定的profile環境變數生效，或者重新登入也可以： ~# source /etc/profile 將撰寫好的Java程式編譯成class檔： ~# javac [程式名稱].java

6 不使用IDE環境(2/2) 在Hadoop上執行程式 (只有一個class檔時)：
若有多個class檔，則需要將class檔包裝成jar檔： ~# jar cvf [jar檔名稱].jar [程式名稱].class 在Hadoop上執行包裝好的jar檔： /hadoop# bin/hadoop jar [jar檔名稱].jar [主函式名稱] [參數0] [參數1] …

7 使用IDE (Eclipse) (1/2) 先到Eclipse官網 ( 下載Eclipse安裝檔 Linux系統上需先安裝圖形化套件 目前下載版本是Eclipse Classic 3.6.2 下載完後解壓縮，並將解壓縮後的目錄移到/opt/eclipse

8 使用IDE (Eclipse) (2/2) 也可以開啟終端機輸入下列指令：
~# wget ~# tar zxvf eclipse-SDK linux-gtk.tar.gz ~# mv eclipse /opt/ 在/usr/local/bin/裡建立eclipse執行檔的連結： ~# ln -sf /opt/eclipse/eclipse /usr/local/bin/ 將/opt/hadoop裡的eclipse plugin搬到eclipse/plugin裡： ~# cp /opt/hadoop/contrib/eclipse-plugin/hadoop eclipse-plugin.jar /opt/eclipse/plugins/ 接下來便可以啟動Eclipse了： ~# eclipse &

9 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

10 新增專案(1/15)

11 新增專案(2/15)

12 新增專案(3/15)

13 新增專案(4/15)

14 新增專案(5/15)

15 新增專案(6/15)

16 新增專案(7/15)

17 新增專案(8/15)

18 新增專案(9/15)

19 新增專案(10/15)

20 新增專案(11/15)

21 新增專案(12/15)

22 新增專案(13/15)

23 新增專案(14/15)

24 新增專案(15/15)

25 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

26 MapReduce程式架構 MapReduce程式主要可分為三個部份 MapReduce Driver Mapper Reducer
利用一個輸入key/value pair集合來產生一個輸出的key/value pair集合 Reducer 接受一個中間key的值和相關的一個value值的集合

27 MapReduce Driver 01. Class MapReduceDriver類別名稱 { 02. main(){
Configuration conf = new Configuration(); Job job = new Job(conf, Job名稱); job.setJarByClass( MapReduceDriver類別(即此類別) ); job.setMapperClass( Mapper類別 ); job.setReducerClass( Reducer類別 ); FileInputFormat.addInputPath( job, new Path(args[0])); FileOutputFormat.setOutputPath( job, new Path(args[1])); 10. 其它參數設定 job.waitForCompletion(true); 12. } 13. }

28 Mapper程式架構 01. class Mapper類別名稱 extends
全域變數 03. public void map( 輸入鍵類型 key, 輸入值類型 value, Context context) throws IOException, InterruptedException { Map程式碼區 context.write(IntermediateKey, IntermediateValue); 06. } 07. }

29 Reducer程式架構 01. class Reducer類別名稱 extends
Redcuer < 輸入鍵類型, 輸入值類型, 輸出鍵類型, 輸出鍵值型 > { 全域變數 03. public void reduce( 輸入鍵類型 key, Iterable< 輸入值類型 > value, Context context) throws IOException, InterruptedException { Reduce程式碼 context.write(ResultKey, ResultValue); 06. } 07. }

30 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

31 MapReduce基礎實作範例(1/2) 本範例以一個簡單的maxCPU程式說明如何使用Eclipse開發MapReduce程式
此範例中，系統每小時記錄一次CPU使用率到日誌檔中，而maxCPU程式會分析日誌檔，並透過MapReduce的方式，找出每天最高的CPU使用率。 本範例中將以這個日誌檔做為輸入檔，先在HDFS上新創一個log目錄，再根據上述格式建立一個日誌檔並上傳到log目錄中。

32 MapReduce基礎實作範例(2/2) 日誌檔中記錄的欄位分別為日期、時段及CPU使用率，目誌檔部份內容如下
2011/01/01 00:00 40 2011/01/01 01:00 30 … 2011/01/02 22:00 40 2011/01/02 23:00 30

33 新增Mapper類別(1/3)

34 新增Mapper類別(2/3) 之後在HadoopLab專案中便會新增一個新的package MR_Lab及mymapper.java ，並修改mymapper.java的內容： 01. public class mymapper extends Mapper<Object, Text, Text, IntWritable> { 02. private Text tday = new Text(); 03. private IntWritable idata = new IntWritable(); 04. public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String day = line.substring(0, 10); String data = line.substring(17); tday.set(day); idata.set(Integer.valueOf(data)); context.write(tday, idata); 11. } 12. }

35 新增Mapper類別(3/3)

36 新增Reducer類別(1/3)

37 新增Reducer類別(2/3) 在MR_Lab的package中出現myreducer.java ，並修改myreducer.java的內容為： 01. public class myreducer extends Reducer<Text, IntWritable, Text, IntWritable> { 02. IntWritable cpuUtil = new IntWritable(); 03. public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int maxValue = Integer.MIN_VALUE; for (IntWritable val : values) { maxValue = Math.max(maxValue, val.get()); } cpuUtil.set(maxValue); context.write(key, cpuUtil); 10. } 11. }

38 新增Reducer類別(3/3)

39 新增MapReduce Driver類別(1/4)

40 新增MapReduce Driver類別(2/4)
在MR_Lab的package中出現maxCPU.java ，並修改maxCPU.java的內容為： 01. public class maxCPU { 02. 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: maxCPU <in> <out>"); System.exit(2); } Job job = new Job(conf, "max CPU"); job.setJarByClass(maxCPU.class); job.setMapperClass(mymapper.class); job.setCombinerClass(myreducer.class); job.setReducerClass(myreducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

41 新增MapReduce Driver類別(3/4)
boolean status = job.waitForCompletion(true); if (status) { System.exit(0); } else { System.err.print("Not Complete!"); System.exit(1); } 25. } 26. }

42 新增MapReduce Driver類別(4/4)

43 在Hadoop上執行MapReduce程式(1/3)

44 在Hadoop上執行MapReduce程式(2/3)

45 在Hadoop上執行MapReduce程式(3/3)
執行結束後，在HDFS上的output目錄中即可看到最後結果： 2011/01/ 2011/01/ 2011/01/ 2011/01/

46 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

47 MapReduce進階實作範例 本範例將所介紹的MapReduce程式 (maxCPU)，加入HDFS及HBase的相關操作。 7.
4. 3. Mapper Reducer HBase 6. 2. 5. HDFS 1. Local host

48 新增maxCPU類別(1/2) 新增一個MapReduce Driver類別並命名為maxCPU，maxCPU類別負責MapReduce相關設定及運作流程 01. public class maxCPU { 02. 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: maxCPU <in> <out>"); System.exit(2); } Job job = new Job(conf, "max CPU"); job.setJarByClass(maxCPU.class); job.setMapperClass(mymapper.class); job.setCombinerClass(myreducer.class); job.setReducerClass(myreducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0]));

49 新增maxCPU類別(2/2) FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); CheckDir.check(otherArgs[0].toString(), conf); LocalToHdfs.localToHdfs(otherArgs[0].toString(), otherArgs[0].toString(), conf); CheckDir.check(otherArgs[1].toString(), conf); CheckTable.check("CPU"); CheckTable.addFamily("CPU", "CPUUtil"); boolean status = job.waitForCompletion(true); if (status) { OutputResult.output(otherArgs[1].toString(), conf); System.exit(0); } else { System.err.print("Not Complete!"); System.exit(1); } 31. } 32. }

50 新增mymapper類別 新增一個Mapper類別並命名為mymapper，其功能為整理輸入的鍵/值，並在第10行呼叫AddData類別將資料存入HBase 01. public class mymapper extends Mapper<Object, Text, Text, IntWritable> { 02. private Text tday = new Text(); 03. private IntWritable idata = new IntWritable(); 04. public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String day = line.substring(0, 10); String time = line.substring(11, 16); String data = line.substring(17); try { AddData.add("CPU", "CPUUtil", day + " " + time, data); } catch (Exception e) { System.err.print("ERROR! (add data to HBase)"); 13. } 14. tday.set(day); 15. idata.set(Integer.valueOf(data)); 16. context.write(tday, idata); 17. } 18. }

51 新增AddData類別 此類別負責將資料插入HBase中 01. public class AddData {
02. public static Configuration configuration = null; 03. static { configuration = HBaseConfiguration.create(); configuration.set("hbase.master", "Host01:60000"); configuration.set("hbase.zookeeper.quorum", "Host01,Host02"); configuration.set("hbase.zookeeper.property.clientPort", "2222"); 08. } 09. static void add(String table, String family, String dtime, String data) throws Exception { HTable htable = new HTable(configuration, table); Put row = new Put(dtime.getBytes()); row.add(family.getBytes(), new String("data").getBytes(), data.getBytes()); htable.put(row); htable.flushCommits(); 15. } 16. }

52 新增myreducer類別 新增一個Reducer類別並命名為myreducer，其程式碼如下
01. public class myreducer extends Reducer<Text, IntWritable, Text, IntWritable> { 02. IntWritable cpuUtil = new IntWritable(); 03. public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int maxValue = Integer.MIN_VALUE; for (IntWritable val : values) { maxValue = Math.max(maxValue, val.get()); } cpuUtil.set(maxValue); context.write(key, cpuUtil); 10. } 11. }

53 新增CheckDir類別 此類別的功能為檢查HDFS上是否有已存在某個檔案或目錄，若有則刪除該檔案或目錄，因此在每次執行程式時，都能自動刪除HDFS上的輸出檔案或目錄 01. public class CheckDir { 02. static void check(final String path, Configuration conf) { Path dstPath = new Path(path); try { FileSystem hdfs = dstPath.getFileSystem(conf); if (hdfs.exists(dstPath)) { hdfs.delete(dstPath, true); } } catch (IOException e) { e.printStackTrace(); } 12. } 13. }

54 新增LocalToHdfs類別 此類別將本地端檔案系統中路徑為src的檔案或目錄上傳至HDFS中路徑為dst的目錄中
01. public class LocalToHdfs { 02. static void localToHdfs(String src, String dst, Configuration conf) { Path dstPath = new Path(dst); try { FileSystem hdfs = dstPath.getFileSystem(conf); hdfs.copyFromLocalFile(false, new Path(src), new Path(dst)); } catch (IOException e) { e.printStackTrace(); } 10. } 11. }

55 新增CheckTable類別(1/2) 此類別中的.check() method與CheckDir類別的功能類似；先檢查HBase中是否有相同的表格，若有則刪除該表格 而.addFamily() method的功能為新增表格並設定其column family。 01. public class CheckTable { 02. public static Configuration configuration = null; 03. static { configuration = HBaseConfiguration.create(); configuration.set("hbase.master", "Host01:60000"); configuration.set("hbase.zookeeper.quorum", "Host01,Host02"); configuration.set("hbase.zookeeper.property.clientPort", "2222"); 08. } 09. public static void check(String table) throws Exception { HBaseAdmin admin = new HBaseAdmin(configuration); if (admin.tableExists(table)) { System.out.println("delete the table "); admin.disableTable(table);

56 新增CheckTable類別(2/2) 14. admin.deleteTable(table); 15. } 16. }
} 16. } 17. public static void addFamily(String table, String family) throws Exception { HBaseAdmin admin = new HBaseAdmin(configuration); HTableDescriptor tableDescripter = new HTableDescriptor(table .getBytes()); tableDescripter.addFamily(new HColumnDescriptor(family)); admin.createTable(tableDescripter); 22. } 23. }

57 新增OutputResult類別(1/2) 此類別由HDFS讀取myreducer的執行結果，得知每日最高CPU使用率，再呼叫ScanTable類別並傳送相關資訊，以找出每日CPU使用率最高的所有時段。 01. public class OutputResult { 02. static void output(final String path, Configuration conf) { Path dst_path = new Path(path); String day = null; String value = null; try { FileSystem hdfs = dst_path.getFileSystem(conf); FSDataInputStream in = null; if (hdfs.exists(dst_path)) { in = hdfs.open(new Path(dst_path.toString() + "/part-r-00000")); String messagein = null; while ((messagein = in.readLine()) != null) { StringTokenizer itr = new StringTokenizer(messagein); day = itr.nextToken(); value = itr.nextToken(); ScanTable.setFilter("CPU", day, value); }

58 新增OutputResult類別(2/2) 18. in.close(); 19. }
} } catch (IOException e) { e.printStackTrace(); } 23. } 24. }

59 新增ScanTable類別(1/2) 類別根據myreducer類別所計算出的每日最高CPU使用率，再到HBase表格中過濾出每日最高使用率的時段並顯示出來 01. public class ScanTable { 02. public static Configuration configuration = null; 03. static { configuration = HBaseConfiguration.create(); configuration.set("hbase.master", "Host01:60000"); configuration.set("hbase.zookeeper.quorum", "Host01,Host02"); configuration.set("hbase.zookeeper.property.clientPort", "2222"); 08. } 09. public static void setFilter(String tablename, String day, String value) throws IOException { HTable table = new HTable(configuration, tablename); Scan scan = new Scan((day + " 00:00").getBytes(), (day + " 23:00").getBytes()); FilterList filterList = new FilterList(); filterList.addFilter(new SingleColumnValueFilter("CPUUtil".getBytes(), "data".getBytes(), CompareOp.EQUAL, value.getBytes())); scan.setFilter(filterList);

60 新增ScanTable類別(2/2) ResultScanner ResultScannerFilterList = table.getScanner(scan); for (Result rs = ResultScannerFilterList.next(); rs != null; rs = ResultScannerFilterList.next()) { for (KeyValue kv : rs.list()) { System.out.println(new String(kv.getRow()) + " " + new String(kv.getValue())); } } 21. } 22. }

61 執行結果 最終的輸出結果如下： 2011/01/01 16:00 100 2011/01/01 17:00 100
2011/01/02 15:00 90 2011/01/03 16:00 80 2011/01/03 17:00 80 2011/01/03 18:00 80 2011/01/04 00:00 40

62 開發環境設定 新增專案 MapReduce程式架構 MapReduce基礎實作範例 MapReduce進階實作範例 Hadoop專題

63 Hadoop專題 專題內容必須包含下列技術： 可考慮加入Hadoop其他子專案技術 MapReduce HDFS Hbase
考慮網路上大量的資料，分別存放在不同的Node，並建立一組HDFS Hbase 利用Hbase過濾出重要資訊，並顯示出來 可考慮加入Hadoop其他子專案技術 Avro、Pig、Hive、Chukwa