1.薪資數據集
我們要寫一個薪資統計程序，統計數據來自於互聯網招聘hadoop崗位的招聘網站，這些數據是按照記錄方式存儲的，因此非常適合使用 MapReduce 程序來統計。
2.數據格式
我們使用的數據來自互聯網招聘網站，其中每一行是一條記錄。
下面我們展示一行數據，其中重要的字段被突出顯示。該行數據被分割成很多行以突出每個字段，但在實際文件中，這些字段被整合成一行
美團
3-5年經驗 # 工作年限
15-30k # 薪資
北京
【夠牛就來】hadoop高級工程
3.分析
在這里，map階段的輸入是原始數據。我們選擇文本格式作為輸入格式，將數據集的每一行作為文本輸入。鍵是某一行起始位置相對於文件起始位置的偏移量，不過我們不需要這個信息，所以將其忽略。
我們的map函數很簡單。由於我們只對工作年限和薪資感興趣，所以只需要取出這兩個字段數據。在本實戰中，map 函數只是一個數據提取階段，通過這種方式來提取數據，使 reducer 函數繼續對它進行處理：即統計不同工作年限的薪資水平
為了全面了解 map 的工作方式，輸入以下數據作為演示
美團 3-5年經驗 15-30k 北京 【夠牛就來】hadoop高級工程...
北信源 3-5年經驗 15-20k 北京 Java高級工程師（有Hadoo...
蘑菇街 3-5年經驗 10-24k 杭州 hadoop開發工程師
晶贊科技 1-3年經驗 10-30k 上海 hadoop研發工程師
秒針系統 3-5年經驗 10-20k 北京 hadoop開發工程師
搜狐 1-3年經驗 18-23k 北京 大數據平台開發工程師（Hadoo...
執御 1-3年經驗 8-14k 杭州 hadoop工程師
KK唱響 3-5年經驗 15-30k 杭州 高級hadoop開發工程師
晶贊科技 1-3年經驗 12-30k 上海 高級數據分析師（hadoop）
億瑪在線 3-5年經驗 18-30k 北京 hadoop工程師
酷訊 1-3年經驗 10-20k 北京 hadoop Engineer/...
游族網絡 5-10年經驗 20-35k 上海 hadoop研發工程師
易車公司 3-5年經驗 15-30k 北京 hadoop工程師
愛圖購 1-3年經驗 8-15k 杭州 hadoop開發工程師
晶贊科技 3-5年經驗 15-33k 上海 hadoop研發工程師
這些數據，以鍵/值對的方式作為map函數的輸入，如下所示
0, 美團 3-5年經驗 15-30k 北京 【夠牛就來】hadoop高級工程...
84, 北信源 3-5年經驗 15-20k 北京 Java高級工程師（有Hadoo...
163, 蘑菇街 3-5年經驗 10-24k 杭州 hadoop開發工程師
231 ,晶贊科技 1-3年經驗 10-30k 上海 hadoop研發工程師
303, 秒針系統 3-5年經驗 10-20k 北京 hadoop開發工程師
375, 搜狐 1-3年經驗 18-23k 北京 大數據平台開發工程師
461, 執御 1-3年經驗 8-14k 杭州 hadoop工程師
521, KK唱響 3-5年經驗 15-30k 杭州 高級hadoop開發工程師
593, 晶贊科技 1-3年經驗 12-30k 上海 高級數據分析師（hadoop）
677, 億瑪在線 3-5年經驗 18-30k 北京 hadoop工程師
774, 酷訊 1-3年經驗 10-20k 北京 hadoop Engineer/...
838, 游族網絡 5-10年經驗 20-35k 上海 hadoop研發工程師
910, 易車公司 3-5年經驗 15-30k 北京 hadoop工程師
976, 愛圖購 1-3年經驗 8-15k 杭州 hadoop開發工程師
1043,晶贊科技 3-5年經驗 15-33k 上海 hadoop研發工程師
鍵(key)是文件中的偏移量，這里不需要這個信息，所以將其忽略。map 函數的功能僅限於提取工作經驗和薪資，並將它們輸出；map 函數的輸出經由 MapReduce 框架處理后，最后發送到reduce函數。這個處理過程基於鍵來對鍵值對進行排序和分組。因此在這個示例中，reduce 函數看到的是如下輸入：
1-3年經驗,[12-30k, 10-30k, 8-14k, 18-23k, 10-20k, 8-15k]
3-5年經驗,[15-33k, 15-30k, 18-30k, 15-30k, 10-20k, 10-24k, 15-20k, 15-30k]
5-10年經驗,[20-35k]
每個工作年限后面緊跟着一系列薪資數據，reduce 函數現在要做的是遍歷整個列表並統計薪資：
1-3年經驗 8-30k
3-5年經驗 10-33k
5-10年經驗 20-35k
上面就是最終輸出結果即統計不同工作年限的薪資水平
4.實現
上面已經分析完畢，下面我們就着手實現它。這里需要編寫三塊代碼內容
1、map 函數、
2、reduce函數
3、一些用來運行作業的代碼。
1、map 函數
下面我們來編寫 Mapper 類，實現 map() 函數，提取工作年限和薪資
1 /* 2 3 * 提取數據 4 5 */ 6 7 public static class SalaryMapper extends Mapper<LongWritable, Text, Text, Text>{ 8 9 public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException 10 11 { 12 13 String[] valArr = value.toString().split("\\s+"); 14 15 16 17 if(valArr.length > 2){ 18 19 /* 20 21 * 美團 3-5年經驗 15-30k 北京 【夠牛就來】hadoop高級工程 22 23 * 24 25 * 查看原始數據，我們只需提取第二字段、第三個字段 26 27 */ 28 29 // 工作年限 30 31 String workAge = valArr[1]; 32 33 // 薪資 34 35 String salary = valArr[2]; 36 37 38 39 context.write(new Text(workAge), new Text(salary)); 40 41 } 42 43 } 44 45 }
這個 Mapper 類是一個泛型類型，它有四個形參類型，分別指定 map 函數的輸入鍵、輸入值、輸出鍵和輸出值的類型。 就本示例來說，輸入鍵是一個長整數偏移量，輸入值是一行文本，輸出鍵是工作年限，輸出值是薪資。Hadoop 本身提供了一套可優化網絡序列化傳輸的基本類型，而不是使用 java 內嵌的類型。這些類型都在 org.apache.hadoop.io 包中。 這里使用 LongWritable 類型（相當於 Java 的 Long 類型）、Text 類型（相當於 Java 中的 String 類型）
map() 方法的輸入是一個鍵（key）和一個值（value），我們首先將 Text 類型的 value 轉換成 Java 的 String 類型， 之后使用 split方法分割數據，然后提取我們需要的值。map() 方法還提供了 Context 實例用於將輸出內容進行寫入。 在這種情況下，我們將工作年限、薪資封裝成Text對象，並將它們進行寫入。
2、reduce函數
下面我們來編寫 Reducer類，實現reduce函數，統計不同工作年限的薪資水平。
1 /* 2 3 * 統計數據 4 5 */ 6 7 public static class SalaryReduce extends Reducer<Text, Text, Text, Text>{ 8 9 public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{ 10 11 // 最低工薪 12 13 int low = 0; 14 15 // 最高薪資 16 17 int hight = 0; 18 19 // 計數功能 20 21 int count = 0; 22 23 24 25 for(Text value : values){ 26 27 String text = value.toString(); 28 29 /* 30 31 * 15-30k 32 33 * 薪資中的分割符是"-" 34 35 */ 36 37 String[] strArr = text.split("-"); 38 39 40 41 int tlow = filterSalary(strArr[0]); 42 43 int thight = filterSalary(strArr[1]); 44 45 46 47 if(count == 0 || low > tlow){ 48 49 low = tlow; 50 51 } 52 53 54 55 if(count == 0 || hight < thight){ 56 57 hight = thight; 58 59 } 60 61 62 63 count ++; 64 65 } 66 67 68 69 context.write(key, new Text(low + "-" + hight + "k")); 70 71 } 72 73 74 75 private int filterSalary(String salary){ 76 77 String sal = Pattern.compile("[^0-9]").matcher(salary).replaceAll(""); 78 79 return Integer.parseInt(sal); 80 81 } 82 83 }
同樣，reduce 函數也有四個形式參數類型用於指定輸入和輸出類型。reduce 函數的輸入類型必須匹配 map 函數的輸出類型：即 Text 類型和Text類型。 在這種情況下，reduce 函數的輸出類型也是 Text 和 Text 類型，分別是工作年限和薪資。在 map 的輸出結果中，所有相同的工作年限（key）被分配到同一個reduce執行，這個薪資就是針對同一個工作年限（key），通過循環所有薪資（values）來求匹配所有薪資的薪資水平。
3、一些用來運行作業的代碼
1 public int run(String[] arg0) throws Exception { 2 3 // 讀取配置文件 4 5 Configuration conf = new Configuration(); 6 7 8 9 // 如果輸出目錄存在，將其刪除 10 11 Path path = new Path(arg0[1]); 12 13 FileSystem fileSystem = path.getFileSystem(conf); 14 15 if(fileSystem.isDirectory(path)){ 16 17 fileSystem.delete(path, true); 18 19 } 20 21 22 23 // 創建Job對象 24 25 Job job = new Job(conf,"salary"); 26 27 job.setJarByClass(Salary.class); 28 29 30 31 // 設置輸入路徑、輸出路徑 32 33 FileInputFormat.addInputPath(job, new Path(arg0[0])); 34 35 FileOutputFormat.setOutputPath(job, new Path(arg0[1])); 36 37 38 39 // 設置mapper、reduce 40 41 job.setMapperClass(SalaryMapper.class); 42 43 job.setReducerClass(SalaryReduce.class); 44 45 46 47 // 設置mapper、reduce的輸出類型 48 49 job.setOutputKeyClass(Text.class); 50 51 job.setOutputValueClass(Text.class); 52 53 54 55 return job.waitForCompletion(true) ? 0 : 1; 56 57 }
1 /** 2 3 * main 方法 4 5 * 6 7 * @param args 8 9 * @throws Exception 10 11 */ 12 13 public static void main(String[] args) throws Exception { 14 15 // 數據輸入路徑和輸出路徑 16 17 String[] args0 = { 18 19 "hdfs://ljc:9000/buaa/salary/", 20 21 "hdfs://ljc:9000/buaa/salary/out/" 22 23 }; 24 25 26 27 int ec = ToolRunner.run(new Configuration(), new Salary(), args0); 28 29 30 31 System.exit(ec); 32 33 }
Configuration 類讀取 Hadoop 的配置文件，如 site-core.xml、mapred-site.xml、hdfs-site.xml 等。
Job 對象指定作業執行規范，我們可以用它來控制整個作業的運行。我們在 Hadoop 集群上運行這個作業時，要把代碼打包成一個JAR文件(Hadoop在集群上發布這個文件)。 不必明確指定 JAR 文件的名稱，在 Job 對象的 setJarByClass 方法中傳遞一個類即可，Hadoop 利用這個類來查找包含它的 JAR 文件，進而找到相關的 JAR 文件。
構造 Job 對象之后，需要指定輸入和輸出數據的路徑。
- 調用 FileInputFormat 類的靜態方法 addInputPath() 來定義輸入數據的路徑，這個路徑可以是單個的文件、一個目錄（此時，將目錄下所有文件當作輸入）或符合特定文件模式的一系列文件。由函數名可知，可以多次調用 addInputPath() 來實現多路徑的輸入。
- 調用 FileOutputFormat 類中的靜態方法 setOutputPath() 來指定輸出路徑（只能有一個輸出路徑）。這個方法指定的是 reduce 函數輸出文件的寫入目錄。 在運行作業前該目錄是不應該存在的，否則 Hadoop 會報錯並拒絕運行作業。這種預防措施的目的是防止數據丟失（長時間運行的作業如果結果被意外覆蓋，肯定是件可怕的事情）。
- 通過 setMapperClass() 和 setReducerClass() 指定 map 類型和reduce 類型。
- 通過setOutputKeyClass() 和 setOutputValueClass() 控制 map 和 reduce 函數的輸出類型，正如本例所示，這兩個輸出類型一般都是相同的。如果不同，則通過 setMapOutputKeyClass()和setMapOutputValueClass()來設置 map 函數的輸出類型。
- 輸入的類型通過 InputFormat 類來控制，我們的例子中沒有設置，因為使用的是默認的 TextInputFormat（文本輸入格式）。
- Job 中的 waitForCompletion() 方法提交作業並等待執行完成。該方法中的布爾參數是個詳細標識，所以作業會把進度寫到控制台。 waitForCompletion() 方法返回一個布爾值，表示執行的成（true）敗（false），這個布爾值被轉換成程序的退出代碼 0 或者 1。
5.結果
6.其他問題
如果，您認為閱讀這篇博客讓您有些收獲，不妨點擊一下右下角的【推薦】。
如果，您希望更容易地發現我的新博客，不妨點擊一下左下角的【關注我】。
如果，您對我的博客所講述的內容有興趣，請繼續關注我的后續博客，我是【劉超★ljc】。
本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須保留此段聲明，且在文章頁面明顯位置給出原文連接，否則保留追究法律責任的權利。