本教程中的代碼分為 3 個部分：

解釋 SalesMapper 類

解釋 SalesCountryReducer 類

解釋 SalesCountryDriver 類

SalesMapper類的說明

在本節中，我們將了解 SalesMapper 類的實現。

我們首先指定類的包名稱。?SalesCountry?就是這個示例中使用的包名。請注意編譯的輸出，SalesMapper.class?將進入目錄并命名這個軟件包名稱：SalesCountry.

其次，我們導入庫軟件包。

以下快照顯示實現 SalesMapper 類?

代碼解釋:

1. SalesMapper 類定義

public class SalesMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {...}

每一個 mapper 類必須從 MapReduceBase 類進行擴展，它必須實現 Mapper 接口。

2. 定義 'map' 函數

1 2 3 4

publicvoidmap(LongWritable key, Text value, OutputCollector<Text, IntWritable> output, Reporter reporter)?throwsIOException

Mapper類的主要部分是接受四個參數的 “map()” 方法。

每次調用?'map()'?方法, 一個鍵值?key-value?對 ('key'?和?'value'?在代碼里) 被傳遞。

'map()'?方法開始被接受拆分輸入文本作為一個參數，并使用分詞來拆分這些行成詞。

1 2	String valueString = value.toString(); String[] SingleCountryData = valueString.split(",");

這里，“,” 被用作分隔符。

在這之后，使用記錄在數組??'SingleCountryData'?中的第七索引，其值為?'1'.

????????output.collect(new Text(SingleCountryData[7]), one);

我們在選擇第7索引記錄，因為我們需要的國家數據，它位于數組?'SingleCountryData' 的第七索引。

請注意，我們輸入的數據是下面的格式 (Country?在索引的位置為：7, ?0 是開始的索引)-

Transaction_date,Product,Price,Payment_Type,Name,City,State,Country,Account_Created,Last_Login,Latitude,Longitude

mapper的輸出使用的是?'OutputCollector'?的?'collect()' 方法的鍵值對.

SalesCountryReducer 類的說明

在本節中，我們將了解 SalesCountryReducer 類的實現。

1. 我們首先為類指定包的名稱。SalesCountry 是包的名稱。請注意編譯的輸出，?SalesCountryReducer.class?將進入命名這個軟件包名稱目錄：?SalesCountry.

其次，我們導入庫軟件包。

以下快照顯示實現 SalesCountryReducer 類

代碼解釋:

1. SalesCountryReducer 類定義 -

public class SalesCountryReducer extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {

此處，前兩個數據類型，?'Text'?和?'IntWritable'?是輸入鍵值的數據類型到reducer。

映射器的輸出的形式<CountryName1, 1>, <CountryName2, 1>.映射器的輸出被輸入到reducer。所以，以配合其數據類型，?Text?和?IntWritable?數據在這里輸入被使用。

最后兩個數據類型，'Text' 和 'IntWritable' 是由 reducer 的鍵 - 值對的形式生成的輸出的數據類型。

每個 reducer 類必須從MapReduceBase類進行擴展，它必須實現 Reducer 接口。

2. Defining 'reduce' function-

1 2 3

publicvoidreduce( Text t_key, Iterator<IntWritable> values,??OutputCollector<Text,IntWritable> output, Reporter reporter)?throwsIOException {

輸入到 reduce() 方法是在具有多個值的列表中選擇一個鍵。

例如，在我們的示例中，這將是 -

<United Arab Emirates, 1>, <United Arab Emirates, 1>, <United Arab Emirates, 1>,<United Arab Emirates, 1>, <United Arab Emirates, 1>, <United Arab Emirates, 1>.

這賦予 reducer 作為?<United Arab Emirates, {1,1,1,1,1,1}>

因此，接受這種形式參數，前兩個數據類型的使用，即 Text 和?Iterator<IntWritable>.?Text是一個數據類型的鍵 和?Iterator<IntWritable>為對于鍵的值的列表的數據類型。

接下來的參數的類型是?OutputCollector<Text,IntWritable>?它收集 reducer 階段的輸出。

reduce()?方法開始通過復制鍵值和初始化頻率計數為0。

????????Text key = t_key;
????????int frequencyForCountry = 0;

然后，使用 “while” 循環，我們通過與鍵關聯的值列表循環，并通過總結所有計算的值。

1 2 3 4 5 6

while(values.hasNext()) { // replace type of value with the actual type of our value IntWritable value = (IntWritable) values.next(); frequencyForCountry += value.get(); }

現在，結果中的鍵得到的頻率計數輸出到收集器。

下面的代碼執行這個 -

????????output.collect(key, new IntWritable(frequencyForCountry));

SalesCountryDriver類的說明

在本節中，我們將了解 SalesCountryDriver 類實現。

1. 我們首先為類指定包的名稱。?SalesCountry?是這里使用的包名。請注意編譯的輸出，?SalesCountryDriver.class?將進入命名這個包名稱的目錄:?SalesCountry.

這里一行指定是包名稱后面的代碼是導入庫軟件包。

2. 定義一個用于創建一個新的客戶端工作，配置 Mapper及Reducer 類對象驅動程序類。

該驅動程序類負責設置我們的 MapReduce 作業在 Hadoop 運行。 在這個類中，我們指定作業名稱，輸入/輸出，mapper 和 reducer 類名稱的數據類型。

3. 在下面的代碼片段中，我們設置這是用來輸入數據集消費和生產輸出，分別輸入和輸出目錄。

arg[0]?和?arg[1]?是通過 MapReduce 的實際操作，也就是賦予在命令行參數執行命令，

$HADOOP_HOME/bin/hadoop jar ProductSalePerCountry.jar /inputMapReduce /mapreduce_output_sales

4. 觸發我們的作業

下面的代碼開始執行 MapReduce 作業

try{// Run the jobJobClient.runJob(job_conf);} catch(Exception e) {e.printStackTrace();}

在MapReduce的計數器是用于收集關于 MapReduce 工作的統計信息的機制。這個信息在MapReduce的作業處理的問題的診斷是很有用的。 計數器類似于將在 map 或 reduce 在代碼日志信息中。

通常情況下，這些計數器在一個程序(map 或 reduce)中定義，當一個特定事件或條件(特定于該計數器)發生執行期間遞增。計數器是一個很好的應用來從輸入數據集跟蹤有效和無效的記錄。

有兩種類型的計數器：

1.?Hadoop?內置計數器:?有一些內置計數器存在每個作業中。下面是內置計數器組：

• MapReduce任務計數器?- 收集任務的具體信息(例如，輸入記錄的數量)在它的執行期間。
• 文件系統計數器?- 收集信息像由一個任務讀取或寫入的字節數
• FileInputFormat計數器?- 收集通過FileInputFormat讀取的字節數的信息
• FileOutputFormat計數器?- 收集的字節數量的信息通過 FileOutputFormat 寫入
• Job?計數器-?這些計數器使用 JobTracker。它們收集統計數據包括如，任務發起了作業的數量。

2. 用戶定義的計數器

除了內置的計數器，用戶可以定義自己的計數器，通過使用編程語言提供了類似的功能。 例如，在 Java 的枚舉用于定義用戶定義的計數器。

一個MapClass例子使用計數器計算缺失和無效值的數量：

publicstaticclassMapClass extendsMapReduceBase implementsMapper<LongWritable, Text, Text, Text> { staticenumSalesCounters { MISSING, INVALID }; publicvoidmap ( LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter)?throwsIOException { //Input string is split using ',' and stored in 'fields' array String fields[] = value.toString().split(",", -20); //Value at 4th index is country. It is stored in 'country' variable String country = fields[4]; //Value at 8th index is sales data. It is stored in 'sales' variable String sales = fields[8]; if(country.length() ==?0) { reporter.incrCounter(SalesCounters.MISSING,?1); }?elseif(sales.startsWith("\"")) { reporter.incrCounter(SalesCounters.INVALID,?1); }?else{ output.collect(newText(country),?newText(sales +?",1")); } } }

上面的代碼片段顯示在 Map Reduce 實現計數器的示例。

在這里，SalesCounters是用“枚舉”定義的計數器。它被用來計算 MISSING 和 INVALID 的輸入記錄。

在代碼段中，如果 “country” 字段的長度為零那么它的值丟失，因此相應的計數器 SalesCounters.MISSING 遞增。

接下來，如果 “sales” 字段開頭是符號 '' ，則記錄被視為無效。這通過遞增計數器 SalesCounters.INVALID 來表示。

MapReduce 連接

連接兩個大的數據集可以使用 MapReduce Join 來實現。然而，這個過程需要編寫大量的代碼來執行實際的連接操作。

連接兩個數據集開始是通過比較每個數據集的大小。如果因為相比其他數據集一個數據集小，那么小數據集被分布到集群中的每個數據節點。一旦分散，無論是 Mapper 或 Reducer 使用更小的數據集進行查找匹配的大型數據集的記錄，然后結合這些記錄，形成輸出記錄。

這取決于在實際連接進行的地方，這個連接分為：

1. 映射端連接 -?當該聯接是由映射器執行的，它稱為映射端鏈接。在這種類型中，聯結前的數據由映射函數實際來消耗的處理。它是強制性的，輸入到每個映射是在分區中的形式，并且是按排序順序。另外，必須有一個相等數目的分區，它必須由連接鍵進行排序。

2. Reduce端連接-?當連接是通過減速器進行的，稱為reduce端連接。沒有必要在此連接有數據集中在以結構化形式(或分區)。

在這里，映射端的處理發出連接這兩個表的關鍵字和對應的元組。作為該處理的效果，所有的元組相同連接鍵都落在相同的 reducer，然后使用相同的連接鍵連接記錄。

整體處理流程示于下圖。

這里有兩個數據集合在兩個不同的文件中，如下所示：

?
?

DEPT_ID 鍵在這兩個文件中常見的。

目標是使用 MapReduce 加入來組合這些文件。

輸入:?我們的輸入數據集是兩個txt文件：DeptName.txt 和 DepStrength.txt

下載輸入文件

前提條件：

• 本教程是在 Linux 上開發 - Ubuntu操作系統
• 已經安裝的Hadoop(本教程使用2.7.1版本)
• Java的開發運行環境已經在系統上安裝(本教程使用的版本是：1.8.0)

在我們開始實際操作之前，使用的用戶 'hduser_'(使用 Hadoop 的用戶)。

yiibai@ubuntu:~$ su hduser_

?

步驟

Step 1)?復制 zip 文件到您選擇的位置

hduser_@ubuntu:/home/yiibai$?cp?/home/yiibai/Downloads/MapReduceJoin.tar.gz?/home/hduser_/ hduser_@ubuntu:/home/yiibai$?ls?/home/hduser_/

操作過程及結果如下：

?

Step 2) 解壓縮ZIP文件，使用以下命令：

hduser_@ubuntu:~$?sudo?tar?-xvf?MapReduceJoin.tar.gz

Step 3) 進入目錄 MapReduceJoin/

hduser_@ubuntu:~$?cd?MapReduceJoin/

Step 4) ?啟動?Hadoop

hduser_@ubuntu:~/MapReduceJoin$?$HADOOP_HOME/sbin/start-dfs.sh hduser_@ubuntu:~/MapReduceJoin$?$HADOOP_HOME/sbin/start-yarn.sh

Step 5) DeptStrength.txt 和 DeptName.txt 用于此項目的輸入文件

這些文件需要使用以下命令 - 復制到 HDFS 的根目錄下，使用以下命令：

hduser_@ubuntu:~/MapReduceJoin$?$HADOOP_HOME/bin/hdfs?dfs?-copyFromLocal?DeptStrength.txt?DeptName.txt?/

Step 6) 使用以下命令 - 運行程序

hduser_@ubuntu:~/MapReduceJoin$?$HADOOP_HOME/bin/hadoop jar MapReduceJoin.jar /DeptStrength.txt /DeptName.txt /output_mapreducejoin

Step 7)

在執行命令后, 輸出文件 (named 'part-00000') 將會存儲在?HDFS目錄 /output_mapreducejoin?

結果可以使用命令行界面可以看到：

hduser_@ubuntu:~/MapReduceJoin$?$HADOOP_HOME/bin/hdfs dfs -cat /output_mapreducejoin/part-00000

結果也可以通過 Web 界面查看(這里我的虛擬機的IP是 192.168.1.109)，如下圖所示：

現在，選擇 “Browse the filesystem”，并瀏覽到 /output_mapreducejoin

打開?part-r-00000

結果如下所示，點擊 Download 鏈接下載：

打開下載后的 文件，結果如下所示：

注：請注意，下一次運行此程序之前，需要刪除輸出目錄 /output_mapreducejoin

$HADOOP_HOME/bin/hdfs dfs -rm -r /output_mapreducejoin

另一種方法是使用不同的名稱作為輸出目錄。

from: http://www.yiibai.com/hadoop/

總結

以上是生活随笔為你收集整理的Hadoop教程（四）：理解MapReduce、MapReduce计数器和连接、MapReduce Hadoop程序连接数据的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。