從wordcount 開始 mapreduce (C++\hadoop streaming模式)

序：終於開始接觸hadoop了，從wordcount開始

1. 采用hadoop streamming模式

 優點：支持C++ pathon shell 等多種語言，學習成本較低，不需要了解hadoop內部結構

 調試方便：cat input | ./map | sort | ./reduce > output

 hadoop 就是提供了一個分布式平台實現了上述腳本的功能，這是一次mapreduce的過程

 一個例子：

#!/bin/bash
source build.env
$hadoop_bin fs -rmr $env_root    
$hadoop_bin fs -mkdir $env_root 
$hadoop_bin fs -copyFromLocal ./txt  $env_root/txt
$hadoop_bin streaming \
        -jobconf mapred.job.name="word count fuck you man~!" \
        -input  $env_root/txt \       //map程序的輸入：cat input | ./map
        -output $env_root/outputxt \  //reduce程序的輸出 ： ./reduce > output
        -mapper "./wordcount_map"\
        -reducer "./wordcount_reducer"\
        -file    ./wordcount_map\
        -file    ./wordcount_reducer\
        -jobconf mapred.job.map.capacity=1000 \
        -jobconf mapred.job.reduce.capacity=1000 \
        -jobconf mapred.child.ulimit=20000000 \
        -jobconf mapred.job.queue.name=ns-webgis \
        -jobconf mapred.job.priority=HIGH \
        -jobconf mapred.map.tasks.speculative.execution=false \
        -jobconf mapred.map.tasks=10 \
        -jobconf mapred.reduce.tasks=2 
if [ $? -ne 0 ]
then
    echo "error~~~~" >&2
    exit -1
fi
$hadoop_bin fs -get $env_root/outputxt .

2. map ：cat input | ./map >> temp

   1）hadoop平台做了什么：

   a.切分文件：把input文件按照一定的策略切割分成若干個小文件

   b.將若干個小文件分別分發到不同節點上

   c. 每個節點上都有一個map程序，然后將任務分發到不同的節點上

   2）自己要寫的wordcount_map要做什么：

   wordcount_map從input中按行進行讀取，然后按照業務邏輯將讀取到的內容拼成 key \t value的形式  ，這個輸出將會作為reduce程序的輸入

   在這里輸出的是 word 1  此處 word是key  1是value

   注意：此處是標准輸出、輸入 std::cout std::cin  in C++

   key與value之間用\t分割，第一個\t之前的值作為key，之后的都作為value 注意：這個key會被hadoop平台用到 平台不關注value值

#include<iostream>
#include<string>
#include<vector>
using namespace std;
void split(string src,vector<string>& dest,string separator)
{
   string str = src;
   string substring;
   string::size_type start = 0, index;

   do
   {
       index = str.find_first_of(separator,start);
       if (index != string::npos)
       {    
         substring = str.substr(start,index-start);
         dest.push_back(substring);
         start = str.find_first_not_of(separator,index);
         if (start == string::npos) return;                                                                         }    
   }while(index != string::npos);
   substring = str.substr(start);
   dest.push_back(substring);
}
void map()
{
    string line;
    vector<string> vec(2);
    while(cin>>line)
    {
        vec.clear();
        split(line,vec," ");
        vector<string>::iterator it=vec.begin();
        for(;it!=vec.end();++it)
        {
            cout<<*it<<"\t"<<"1"<<"\t"<<"fuck"<<endl;
        }
    }
}
int main()
{
    map();
}

   3. reduce: sort  |  ./reduce > output

   等到所有的map任務都結束:

   1）hadoop平台做了這些事情

        a.將所有的map程序的輸出結果中key相同的key value pair 放到相同的節點上，這點很重要，這是保證最終輸出結果正確的保證，后面會按照key進行hash , 並且相同                 key之間不會有其他的key,其實是按照key值做了一個排序

          注意：相同的key一定在一個節點上，但是一個節點上不止有一個個key

        b 然后在各個節點上開始reduce任務

   2）自己寫的wordcount_map做了什么

        a. 讀取這些具有相同key的鍵值對，處理自己的業務邏輯，此處就是將統計相同的key值的鍵值對一共出現了幾次，然后將結果輸出，此處也是標准輸入和標准輸出

        

#include<vector>
#include<map>
#include<string>
#include<iostream>
using namespace std;
void reduce()
{
    string key;
    string value;
    string value1;
    //vector<string> vec(2);
    map<string,int> mapTemp;
    while(cin>>key>>value>>value1)
    {        
        if(mapTemp.find(key)!=mapTemp.end())
            mapTemp[key]+=1;
        else
            mapTemp[key]=1;
    }
    map<string,int>::iterator it = mapTemp.begin();
    for(;it!=mapTemp.end();++it)
    {
        cout<<it->first<<"\t"<<it->second<<endl;
    }
}
int main()
{
    reduce();
}