Day1-hadoop_ _v1

Transcription

1 王耀聰陳威宇

2 課程大綱 (1) 第一天 09:30~10:20 介紹課程與 Hadoop 簡介 10:20~10:30 休息 10:30~12:00 Hadoop 生態系簡介實作一 :Hadoop Streaming 範例操作 12:00~13:00 午餐 13:00~15:00 開發輔助工具 Eclipse Map Reduce 程式架構 15:00~15:10 休息 15:10~16:30 程式設計 I- HDFS 操作程式設計 II- 範例程式 2

3 課程大綱 (2) 第二天 09:30~10:20 HBase 簡介與架構 10:20~10:30 休息 10:30~12:00 HBase 安裝操作說明 12:00~13:00 午餐 13:00~15:00 HBase 程式架構與範例 15:00~15:10 休息 15:10~16:00 Hadoop + HBase + PHP 案例實務 16:00~16:30 hadoop + 關聯式資料庫 3

4 學員背景調查 Java 語言?? PHP 語言?? MySQL 資料庫?? Linux 操作?? 電腦叢集維護?? 安裝過 Hadoop?? 參加過 Hadoop 基礎課程?? 4

5 It s Show Time 名稱 服務公司 / 就讀學校 報名原因 預期收穫 5

6 Hadoop 程式設計 引言 雲端運算這個名詞雖然紅, 但我一定需要雲端運算嗎? 他用在什麼場合? 又或非它不可嗎? 6

7 Hadoop 程式設計 Computing with big datasets is a fundamentally different challenge than doing big compute over a small dataset 7

8 平行分散式運算 格網運算 ( 網格運算, Grid computing) MPI, PVM, Condor 著重於 : 分散工作量 目前的問題在於 : 如何分散資料量 Reading 100 GB off a single filer would leave nodes starved just store data locally 8

9 分散大量資料 : Slow and Tricky 交換資料需同步處理 Deadlock becomes a problem 有限的頻寬 Failovers can cause cascading failure 9

10 數字會說話 Data processed by Google every month: 400 PB in 2007 Max data in memory: 32 GB Max data per computer: 12 TB Average job size: 180 GB 光一個 device 的讀取時間 = 45 minutes 10

11 所以 運算資料可以很快速, 但瓶頸在於 硬碟的 I/O 1 HDD = 75 MB/sec 解法 : parallel reads 1000 HDDs = 75 GB/sec 11

12 MapReduce 的動機 Data > 1 TB 交互運算於大量的 CPU 容易開發與使用 High-level applications written in MapReduce Programmers don t worry about socket(), etc. 12

13 Conclusions 大資料集的運算 與 小資料集的高速運 算,, 兩者是迥然不同的挑戰 大資料量的解決方法將需要 : 提供囊括所有解決之道的新工具 MapReduce 與 HDFS 等工具是其中之一

14 一 Hadoop 簡介 Hadoop 是一套儲存並處理 petabytes 等級資訊的 雲端運算技術 14

15 Hadoop 以 Java 開發 自由軟體 上千個節點 Petabyte 等級的資料量 創始者 Doug Cutting 為 Apache 軟體基金會的 top level project 15

16 巨量 特色 擁有儲存與處理大量資料的能力 經濟 由一般個人電腦所架設的叢集環境 效率 藉由平行分散檔案以致得到快速的回應 可靠 當某個節點發生錯誤, 系統能即時自動的取得備份資料以及佈署運算資源 16

17 實用案例 Hadoop 於 Yahoo 的運作資訊 年年 日日 節節節 耗耗 ( 小耗 ) 2006 四四 五四 十十四 十十四 十十四 十十四 七四 七四 七四 七四 Sort benchmark, every nodes with terabytes data. 17

18 Facebook 誰在用 Hadoop?(1) 處理 internal log and dimension data sources for reporting/analytics and machine learning. IBM Blue Cloud Computing Clusters Journey Dynamics 用 Hadoop MapReduce 分析 billions of lines of GPS data 並產生交通路線資訊. Krugle 用 Hadoop and Nutch 建構原始碼搜尋引擎 18

19 誰在用 Hadoop?(2) SEDNS - Security Enhanced DNS Group 收集全世界的 DNS 以探索網路分散式內容. Technical analysis and Stock Research 分析股票資訊 University of Nebraska Lincoln, Research Computing Facility 用 Hadoop 跑約 200TB 的 CMS 經驗分析 緊湊渺子線圈 (CMS,Compact Muon Solenoid) 為瑞士歐洲核子研究組織 CERN 的大型強子對撞器計劃的兩大通用型粒子偵測器中的一個 19

20 誰在用 Hadoop? (3) Yahoo! Used to support research for Ad Systems and Web Search 使用 Hadoop 平台來發現發送垃圾郵件的殭屍網絡 趨勢科技 過濾像是釣魚網站或惡意連結的網頁內容 20

21 Hadoop 的主要架構 21

22 Building Hadoop Namenode Hadoop JobTracker Data Task Data Task Data Task Java Java Java Linuux Linuux Linuux 22 Node1 Node2 Node3 22

23 名詞 Job Task 任務 小工作 JobTracker 任務分派者 TaskTracker Client Map 小工作的執行者 發起任務的客戶端 應對 Reduce 總和 Namenode 名稱節點 Datanode 資料節點 Namespace 名稱空間 Replication 副本 Blocks Metadata 檔案區塊 (64M) 屬性資料 23

24 管理資料 Namenode Master 管理 HDFS 的名稱空間 控制對檔案的讀 / 寫 配置副本策略 對名稱空間作檢查及紀錄 只能有一個 Datanode Workers 執行讀 / 寫動作 執行 Namenode 的副本策略 可多個 24

25 分派程序 Jobtracker Master 使用者發起工作 指派工作給 Tasktrackers 排程決策 工作分配 錯誤處理 只能有一個 Tasktrackers Workers 運作 Map 與 Reduce 的工作 管理儲存 回覆運算結果 可多個 25

26 不在雲裡的 Client 26

27 其他的 Open Source 專案 : Sector The National Center for Data Mining (NCDM) 其他不同語言實作的 MapReduce 函式庫

28 關於 Sector / Sphere 由美國資料探勘中心 (National Center for Data Mining) 研發的自由軟體專案 採用 C/C++ 語言撰寫, 因此效能較 Hadoop 更好 提供 類似 Google File System 與 MapReduce 的機制 基於 UDT 高效率網路協定來加速資料傳輸效率 Open Cloud Consortium 的 Open Cloud Testbed, 有提供測試環境, 並開發了 MalStone 效能評比軟體

29 Conclusions 所有工作都由 JobTracker 統一分派, 由眾多 TaskTracker 執行, 每個 TaskTracker 又可以 執行多個 Task threads 所有名稱空間與檔案的 metadata 都由一個 Namenode 統籌, 檔案空間為所有 Datanode 的集合,hdfs 的基本單位為 block Client 只需要丟工作或存取在 雲 的資料 29

30 Hadoop 只支援用 Java 開發嘛? Is Hadoop only support Java? 總不能全部都重新設計吧? 如何與舊系統相容? Can Hadoop work with existing software? 可以跟資料庫結合嘛? Can Hadoop work with Databases? 開發者們有聽到大家的需求... Yes, we hear the feedback of developers... 30

31 Hadoop 程式設計 二 Hadoop 相關子專案 Hadoop Common: The common utilities that support the other Hadoop subprojects. HDFS: A distributed file system that provides high throughput access to application data. MapReduce: A software framework for distributed processing of large data sets on compute clusters. 31

32 Hadoop 相關子專案 Chukwa: A data collection system for managing large distributed systems. HBase: A scalable, distributed database that supports structured data storage for large tables. Hive: A data warehouse infrastructure that provides data summarization and ad hoc querying. Pig: A high-level data-flow language and execution framework for parallel computation. ZooKeeper: A high-performance coordination service 32 for distributed applications.

33 Hadoop 生態系 (Ecosystem) Pig Chukwa Hive HBase MapReduce HDFS ZooKeeper Hadoop Core (Hadoop Common) Avro Source: Hadoop: The Definitive Guide 33

34 Avro Avro is a data serialization system. It provides: Rich data structures. A compact, fast, binary data format. A container file, to store persistent data. Remote procedure call (RPC). Simple integration with dynamic languages. For more detail, please check the official document: 34

35 Zoo Keeper ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services. All of these kinds of services are used in some form or another by distributed applications. 35

36 Pig Pig is a platform for analyzing large data sets that consists of a high-level language for expressing data analysis programs, coupled with infrastructure for evaluating these programs. Pig's infrastructure layer consists of a compiler that produces sequences of Map-Reduce programs Pig's language layer currently consists of a textual language called Pig Latin, which has the following key properties: Ease of programming Optimization opportunities Extensibility

37 Hive Hive is a data warehouse infrastructure built on top of Hadoop that provides tools to enable easy data summarization, adhoc querying and analysis of large datasets data stored in Hadoop files. Hive QL is based on SQL and enables users familiar with SQL to query this data.

38 HBase HBase is a distributed column-oriented database built on top of HDFS. A distributed data store that can scale horizontally to 1,000s of commodity servers and petabytes of indexed storage. Designed to operate on top of the Hadoop distributed file system (HDFS) or Kosmos File System (KFS, aka Cloudstore) for scalability, fault tolerance, and high availability. Integrated into the Hadoop map-reduce platform and paradigm.

39 Chukwa Chukwa is an open source data collection system for monitoring large distributed systems. built on top of HDFS and Map/Reduce framework includes a flexible and powerful toolkit for displaying, monitoring and analyzing results to make the best use of the collected data.

40 Mahout Mahout is a scalable machine learning libraries. implemented on top of Apache Hadoop using the map/reduce paradigm. Mahout currently has Collaborative Filtering User and Item based recommenders K-Means, Fuzzy K-Means clustering Mean Shift clustering More...

41 Hadoop 只支援 Java 嗎? Although the Hadoop framework is implemented in Java TM, Map/Reduce applications need not be written in Java. Hadoop Streaming is a utility which allows users to create and run jobs with any executables (e.g. shell utilities) as the mapper and/or the reducer. Hadoop Pipes is a SWIG-compatible C++ API to implement Map/Reduce applications (non JNI TM based).

42 Hadoop Pipes (C++, Python) Hadoop Pipes allows C++ code to use Hadoop DFS and map/reduce. The C++ interface is "swigable" so that interfaces can be generated for python and other scripting languages. For more detail, check the API Document of org.apache.hadoop.mapred.pipes You can also find example code at hadoop-*/src/examples/pipes About the pipes C++ WordCount example code: 42

43 Hadoop Streaming Hadoop Streaming is a utility which allows users to create and run Map-Reduce jobs with any executables (e.g. Unix shell utilities) as the mapper and/or the reducer. It's useful when you need to run existing program written in shell script, perl script or even PHP. Note: both the mapper and the reducer are executables that read the input from STDIN (line by line) and emit the output to STDOUT. For more detail, check the official document of Hadoop Streaming

44 Running Hadoop Streaming hadoop jar hadoop-streaming.jar help Usage: $HADOOP_HOME/bin/hadoop [--config dir] jar \ $HADOOP_HOME/hadoop-streaming.jar [options] Options: -input <path> DFS input file(s) for the Map step -output <path> DFS output directory for the Reduce step -mapper <cmd JavaClassName> The streaming command to run -combiner <JavaClassName> Combiner has to be a Java class -reducer <cmd JavaClassName> The streaming command to run -file <file> File/dir to be shipped in the Job jar file -dfs <h:p> local Optional. Override DFS configuration -jt <h:p> local Optional. Override JobTracker configuration -additionalconfspec specfile Optional. -inputformat TextInputFormat(default) SequenceFileAsTextInputFormat JavaClassNam e Optional. -outputformat TextOutputFormat(default) JavaClassName Optional. More

45 Hadoop Streaming 範例 (1) hadoop:~$ hadoop fs -rmr input output hadoop:~$ hadoop fs -put /etc/hadoop/conf input hadoop:~$ hadoop jar hadoop-streaming.jar -input input -output output -mapper /bin/cat -reducer /usr/bin/wc 45

46 Hadoop Streaming 範例 (2) hadoop:~$ echo "sed -e \"s/ /\n/g\" grep." > streamingmapper.sh hadoop:~$ echo "uniq -c awk '{print \$2 \"\t\" \$1}'" > streamingreducer.sh hadoop:~$ chmod a+x streamingmapper.sh hadoop:~$ chmod a+x streamingreducer.sh hadoop:~$ hadoop fs -put /etc/hadoop/conf input hadoop:~$ hadoop jar hadoop-streaming.jar -input input -output output -mapper streamingmapper.sh -reducer streamingreducer.sh -file streamingmapper.sh -file streamingreducer.sh 46

47 Hadoop 程式設計 三 Map Reduce 與 HDFS 3.A:HDFS 檔案系統 3.B:MapReduce 演算法 47

48 Hadoop 程式設計 三 Map Reduce 與 HDFS 3.A:HDFS - 檔案系統 HDFS = Hadoop Distributed File System, 是 Hadoop 用來存放資料的分散式檔案系統, 因此若要讓 Hadoop 運算資料, 就要將待運算的資料放到這個空間才可以 ; 同理可知運算後的資料... 48

49 HDFS: 開發動機 基於 Google File System (GFS) 用便宜而不可靠的電腦, 打造互為備援 的儲存媒介, 拿來存放海量資料 為何不使用現存的檔案系統呢? 不同的工作負載與設計優先權 須處理比其他檔案系統還要更大的資料集 49

50 基本假設 高單元故障率 便宜的個人電腦商品總是容易故障 海量檔案的 客觀 參考數字 就只是幾百萬個檔案而已... 每個大小約 100MB 或更大 ; 通常以數 GB 的檔案最常見 這些檔案都只寫一次 (write-once), 多數是附加 (append) 大量的串流讀取需求 持續而大量的資料通量 (throughput) 遠比較短的反應延遲 ( latency) 來得重要 50

51 HDFS 設計準則 檔案以區塊 (block) 方式儲存 每個區塊大小遠比多數檔案系統都來得大 ( 預設值為 64MB) 透過複本機制來提高可靠度 每個區塊至少備分到三台以上的 DataNode 單一 master (NameNode) 來協調存取及屬性資料 (metadata) 簡易的集中控管機制 沒有資料快取機制 (No data caching) 快取對於大資料集與串流讀取沒太大幫助 熟悉的介面, 但客制化的 API 簡化問題 ; 專注於分散式應用 51

52 管理資料 52

53 HDFS 運作 Namenode (the master) 檔案路徑 副本數, 由哪幾個 block 組成 name:/users/joeyahoo/myfile - copies:2, blocks:{1,3} name:/users/bobyahoo/somedata.gzip, copies:3, blocks:{2,4,5} Client Metadata I/O Datanodes (the slaves)

54 HDFS 運作 目的 : 提高系統的可靠性與讀取的效率 可靠性 : 節點失效時讀取副本已維持正常運作 讀取效率 : 分散讀取流量 ( 但增加寫入時效能瓶頸 ) Namenode JobTracker file1 (1,3) file2 (2,4,5) Map tasks Reduce tasks TaskTracker TT ask for task Block 1 1 TT 2 TT TT TT 3 TT

55 HDFS 系統流程圖 55

56 Hadoop 程式設計 三 Map Reduce 與 HDFS 3.B : MapReduce 演算法 Hadoop 的運算方式是透過 Map/Reduce 演算法構成, 也意謂著, 要透過 Hadoop API 撰寫平行分散式程式, 則一定要懂 Map / Reduce 56

57 Map / Reduce 定義 MapReduce is a software framework to support distributed computing on large data sets on clusters of computers. 57

58 演算法 Functional Programming : Map Reduce map(...) : [ 1,2,3,4 ] (*2) -> [ 2,4,6,8 ] reduce(...): [ 1,2,3,4 ] - (sum) -> 10 對應演算法中的 Divide and conquer 將問題分解成很多個小問題之後, 再做總和 58 58

59 Map One-to-one Mapper let map(k, v) = Emit(k.toUpper(), v.toupper()) Explode Mapper let map(k, v) = foreach char c in v: emit(k, c) ( Foo, other ) ( FOO, OTHER ) ( key2, data ) ( KEY2, DATA ) ( A, cats ) ( A, c ), ( A, a ), ( A, t ), ( A, s ) Filter Mapper let map(k, v) = if (isprime(v)) then emit(k, v) ( foo, 7) ( foo, 7) ( test, 10) (nothing) 59

60 Reduce Example: Sum Reducer let reduce(k, vals) = sum = 0 foreach int v in vals: sum += v emit(k, sum) ( A, [42, 100, 312]) ( A, 454) ( B, [12, 6, -2]) ( B, 16) 60

61 MapReduce 單台 Map 數量依輸入檔案的 block 數為主,Reduce 數量由系統決定 Sort 的規則為 key 的字母順序 61 61

62 MapReduce 多台 62 62

63 MapReduce 運作流程 input HDFS map sort/copy merge output HDFS split 0 split 1 reduce part0 split 2 split 3 map split 4 reduce part1 map JobTracker 跟 NameNode 取得需要運算的 blocks JobTracker 選數個 TaskTracker 來作 Map 運算, 產生些中間檔案 JobTracker 將中間檔案整合排序後, 複製到需要的 TaskTracker 去 JobTracker 派遣 TaskTracker 作 reduce reduce 完後通知 JobTracker 與 Namenode 以產生 output 63 63

64 實例範例 I am a tiger, you are also a tiger map map map I,1 am,1 a,1 tiger,1 you,1 are,1 also,1 a, 1 tiger,1 a, 1 a,1 also,1 am,1 are,1 I,1 tiger,1 tiger,1 you,1 reduce reduce a,2 also,1 a,2 am,1 also,1 are,1 am,1 are,1 I,1 tiger,2 you,1 I, 1 tiger,2 you,1 JobTracker 先選了三個 Tracker 做 map Map 結束後,hadoop 進行中間資料的整理與排序 JobTracker 再選兩個 TaskTracker 作 reduce 64

65 Hadoop 程式設計 四 寫 Code 環境準備 4.A : Console 端 編譯與執行 4.B : 透過 Eclipse 開發與測試運算 65

66 Java 之編譯與執行 1. 編譯 javac -classpath hadoop-*-core.jar -d MyJava MyCode.java 2. 封裝 jar -cvf MyJar.jar -C MyJava. 3. 執行 bin/hadoop jar MyJar.jar MyCode HDFS_Input/ HDFS_Output/ 所在的執行目錄為 Hadoop_Home./MyJava = 編譯後程式碼目錄 Myjar.jar = 封裝後的編譯檔 先放些文件檔到 HDFS 上的 input 目錄./input;./ouput = hdfs 的輸入 輸出目錄 66 66

67 WordCount1 練習 (I) 1. cd $HADOOP_HOME; mkdir input_local 2. echo "I like NCHC Cloud Course." > input_local/input1 3. echo "I like nchc Cloud Course, and we enjoy this crouse." > input_local/input2 4. bin/hadoop dfs -put input_local input 5. bin/hadoop dfs -ls input 67

68 WordCount1 練習 (II) 1. 編輯 WordCount.java 2. mkdir MyJava 3. javac -classpath hadoop-*-core.jar -d MyJava WordCount.java 4. jar -cvf wordcount.jar -C MyJava. 5. bin/hadoop jar wordcount.jar WordCount input/ output/ 所在的執行目錄為 Hadoop_Home( 因為 hadoop-*-core.jar ) javac 編譯時需要 classpath, 但 hadoop jar 時不用 wordcount.jar = 封裝後的編譯檔, 但執行時需告知 class name Hadoop 進行運算時, 只有 input 檔要放到 hdfs 上, 以便 hadoop 分析運算 ; 執行檔 (wordcount.jar wordcount.jar) 不需上傳, 也不需每個 node 都放, 程式的載入交由 java 處理 68

69 WordCount1 練習 (III) 69

70 WordCount1 練習 (IV) 70

71 BTW 雖然 Hadoop 框架是用 Java 實作, 但 Map/Reduce 應用程序則不一定要用 Java 來寫 Hadoop Streaming : 執行作業的工具, 使用者可以用其他語言 ( 如 :PHP) 套用到 Hadoop 的 mapper 和 reducer Hadoop Pipes:C++ API 71

72 Hadoop 程式設計 四 寫 Code 環境準備 4.A : Console 端編譯與執行 4.B : 透過 Eclipse 開發與測試運算 72

73 Requirements Hadoop up Java 1.6 Eclipse 3.3 up Hadoop Eclipse Plugin up 73

74 安裝 Hadoop Eclipse Plugin Hadoop Eclipse Plugin Compiler needed Hadoop Eclipse Plugin From $Hadoop_0.20.0_home/contrib/eclipseplugin/hadoop eclipse-plugin.jar Or download from plugin.googlecode.com/files/hadoop eclipse- plugin.jar copy to $Eclipse_home/plugins/ 74

75 1 打開 Eclipse, 設定專案目錄 75

76 2. 使用 Hadoop mode 視野 1. Window 2. Open Perspective 3. Other 若有看到 MapReduce 的大象圖示代表 Hadoop Eclipse plugin 有安裝成功, 若沒有請檢查是否有安之裝正確 76

77 3. 使用 Hadoop 視野, 主畫面將出現三個功能 77

78 4. 建立一個 Hadoop 專案 開出新專案 選擇 Map/Reduce 專案 78

79 4-1. 輸入專案名稱並點選設定 Hadoop 安裝路徑 由此設定專案名稱 由此設定 Hadoop 的安裝路徑 79

80 填入 Hadoop 安裝路徑 於此輸入您 Hadoop 的安裝路徑, 之後選擇 ok 80

81 5. 設定 Hadoop 專案細節 1. 右鍵點選 2. 選擇 Properties 81

82 5-1. 設定原始碼與文件路徑 選擇 Java Build Path 以下請輸入正確的 Hadoop 原始碼與 API 文件檔路徑, 如 source :/opt/hadoop/src/ javadoc:file:/opt/hadoop/docs/api/ 82

83 完成圖 83

84 5-2. 設定 java doc 的完整路徑 選擇 Javadoc Location 輸入 java 6 的 API 正確路徑, 輸入完後可選擇 validate 以驗證是否正確 84

85 6. 連結 Hadoop Server 與 Eclipse 點選此圖示 85

86 6-1. 設定你要連接的 Hadoop 主機 任意填一個名稱 輸入主機位址或 domain name MapRedu ce 監聽的 Port ( 設定於 mapredsite.xml) HDFS 監聽的 Port ( 設定於 coresite.xml) 你在此 Hadoop Server 上的 Username 86

87 6-2 若正確設定則可得到以下畫面 HDFS 的資訊, 可直接於此操作檢視 新增 上傳 刪除等命令 若有 Job 運作, 可於此視窗檢視 87

88 7. 新增一個 Hadoop 程式 首先先建立一個 WordCount 程式, 其他欄位任意 88

89 7.1 於程式窗格內輸入程式碼 此區為程式窗格 89

90 7.2 補充 若之前 doc 部份設定正確, 則滑鼠移至程式碼可取得 API 完整說明 90

91 8. 運作 於欲運算的程式碼處點選右鍵 Run As Run on Hadoop 91

92 8-1 選擇之前設定好所要運算的主機 92

93 8.2 運算資訊出現於 Eclipse 右下方的 Console 視窗 放大 93

94 8.3 剛剛運算的結果出現如下圖 放大 94

95 Conclusions 優點 快速開發程式 易於除錯 智慧尋找函式庫 自動鍊結 API 直接操控 HDFS 與 JobTracker 缺點 Plugin 並會因 Eclipse 版本而有不同的狀況 95

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

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

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

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

100 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

101 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

102 Class Reducer (v 0.20) import org.apache.hadoop.mapreduce.reducer; 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

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

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

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

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

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

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

109 傳送檔案至 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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

126 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

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

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

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

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

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

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

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

134 透過一些使用 Hadoop 的案例帶給大家一些新的啟發

135 A: : 用 Hadoop 打造 Location Plus 2009/11/17 Location Plus! 服務, 擊敗了其他 38 組作品獲得 Yahoo 開發競賽的優勝 從大量的批踢踢 BBS 文章中, 找出臺灣 17 個城市的熱門話題 每個城市提供 30 個熱門話題和相關的參考詞 可以讓使用者用這些熱門話題來搜尋 Yahoo 知識 + 生活 + 無名小站等內容 提供了手機版介面, 讓使用者到任何地方就知道當地有哪些熱門話題 135

136 A:Location Plus 用 RSS 蒐集批踢踢 BBS 站的十大熱門看板的文章, 約 20 萬筆文章記錄 維基百科提供的 30 萬筆詞條作為關鍵詞 將所有關鍵詞套用到所有文章紀錄, 共需 600 億次比對 ( 還不包含排序..)) 這些都交給 Hadoop 吧! 136

137 B:Yahoo 使用 Hadoop 平台來發現發送垃圾郵件的殭屍網絡 (1) 垃圾郵件網站 (2) 垃圾郵件 (3) 垃圾郵件軟體 (4) 被感染的電腦 (5) 病毒或木馬 (6) 信件伺服器 (7) 用戶 (8) 網路流量 137

138 C: : 警訊整合系統 目的 : 將原本複雜難懂的警訊日誌整合成易於明瞭的報告 透過 雲端 來運算大量資料 環境 : hadoop 0.20 Java 1.6 Apache 2 138

139 輸入資料 [**] [1:538:15] NETBIOS SMB IPC$ unicode share access [**] [Classification: Generic Protocol Command Decode] [Priority: 3] 09/04-17:53: :1051 -> :139 TCP TTL:128 TOS:0x0 ID:4000 IpLen:20 DgmLen:138 DF ***AP*** Seq: 0x2E589B8 Ack: 0x642D47F9 Win: 0x4241 TcpLen: 20 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: :1032 -> :1900 UDP TTL:1 TOS:0x0 ID:80 IpLen:20 DgmLen:161 Len: 133 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: :1032 -> :1900 UDP TTL:1 TOS:0x0 ID:82 IpLen:20 DgmLen:161 Len: 133 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: :1032 -> :1900 UDP TTL:1 TOS:0x0 ID:84 IpLen:20 DgmLen:161 Len: 133 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: : > :1900 UDP TTL:1 TOS:0x0 ID:105 IpLen:20 DgmLen:161 Len: 133 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: : > :1900 UDP TTL:1 TOS:0x0 ID:117 IpLen:20 DgmLen:160 Len: 132 [**] [1:1917:6] SCAN UPnP service discover attempt [**] [Classification: Detection of a Network Scan] [Priority: 3] 09/04-17:53: : > :1900 UDP TTL:1 TOS:0x0 ID:118 IpLen:20 DgmLen:161 Len: 133 [**] [1:2351:10] NETBIOS DCERPC ISystemActivator path overflow attempt little endian unicode [**] [Classification: Attempted Administrator Privilege Gain] [Priority: 1] 09/04-17:53: : > :135 TCP TTL:105 TOS:0x0 ID:49809 IpLen:20 DgmLen:1420 DF ***A**** Seq: 0xF9589BBF Ack: 0x82CCF5B7 Win: 0xFFFF TcpLen: 20 [Xref => => => => 139

140 輸出資料 Generate dot graph format 140

141 系統分析 前處理 輸入資料 Map Reduce 運算 後續處理 網頁介面 瀏覽結果報告 141

142 Alert Merge Example Destination IP Attack Signature Source IP Destination Port Source Port Packet Protocol Timestamp Host_1 Trojan Sip tcp T1 Host_1 Trojan Sip tcp T2 Host_1 Trojan Sip tcp T3 Host_2 Trojan Sip tcp T4 Host_3 D.D.O.S Sip udp T5 Host_3 D.D.O.S Sip tcp T5 Host_3 D.D.O.S Sip udp T5 Host_3 D.D.O.S Sip tcp T5 Key Values Host_1 Trojan Sip1,Sip2 80, ,5002 tcp T1,T2,T3 Host_2 Trojan Sip tcp T4 Host_3 D.D.O.S. Sip3,Sip4,Sip5,Sip ,6008 tcp, udp T5 142

143 程式流程圖 Snort Logs Source Regulation Integrate Alert Final Report Correlation Record Merge Extenstion 143

144 Conclusions 評估 系統分析 輸入輸出 系統元件 各元件參數與串流 實做 多次運算 前處理與後處理 144

145