理解 MapReducer


本教學中的程式碼分為 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 job
           JobClient.runJob(job_conf);
       } catch(Exception e) {
           e.printStackTrace();
       }