Mapper開發

由於是透過Standard Input/Output，因此console非常適合，所以會建立一個Mapper的.Net Core Console程式。

在Mapper的階段，內容會是一行一行讀進來，所以把讀進來的內容做文字切割， 每找到一個word，就寫到output，word是key，1是value（代表找到一筆）

會一直迴圈的讀，直到沒有任何檔案為止。如果把這個和之前java比照會發現邏輯一樣。

class Program
{
    static void Main(string[] args)
    {
        string line;

        while ((line = Console.ReadLine()) != null)
        {
            // 用文字切割
            var words = Regex.Matches(line, @"[\w]+");

            foreach (var word in words)
            {
                // 每一個找到的算1筆 - keyvalue用tab切割
                Console.WriteLine("{0}\t1", word);
            }
        }
    }
}

Reducer開發

會在建立另外一個專案用來放Reducer的程式。

Reducer一樣是讀Input然後寫到output。由於這次讀到的內容是從Mapper來的，所以會先用tab做切割，key是word，value就是筆數（也都是1）。

在這邊，有建立一個words dictionary，這個是因為在Mapper階段其實沒有管word有沒有重複，反正出現就是+1。

不過在Reducer因為要加總，因此用了words dictionary作為一個暫存的空間。

最後把所有結果寫到output - 也是 key value pair，key一樣是word，不過value就是word出現的總數。

static void Main(string[] args)
{
    // 用來儲存已經出現過的字 - java版本會自動處理，不過這個stream需要手動記錄
    Dictionary<string, int> words = new Dictionary<string, int>();

    string line;

    while ((line = Console.ReadLine()) != null)
    {
        // 傳過來的key value用tab分割（Mapper也是用tab切割key和value）
        var keyValuePair = line.Split('\t');

        string word = keyValuePair[0];

        int count = Convert.ToInt32(keyValuePair[1]);

        // 如果已經有這個word，和字典的加總，不然就建立新的
        if (words.ContainsKey(word))
        {
            words[word] += count;
        }
        else
        {
            words.Add(word, count);
        }
    }

    // 把所有結果寫出來
    foreach (var word in words)
    {
        Console.WriteLine("{0}\t{1}", word.Key, word.Value);
}

測試結果

當整個程式準備好了之後，接下來就可以對這個程式做測試了。

在接下來將會用一個docker版本的hadoop做測試 - 希望透過docker方式也可以了解用docker做測試有多方便。

接下來的測試都是在powershell可以直接執行。

git clone https://github.com/alantsai/blog-data-science-series.git
cd .\blog-data-science-series\src\chapter-10-dotnet-mapreduce
                        

先把.net core的console 發佈出來

在powershell執行指令：dotnet publish -o ${pwd}\dotnetmapreduce .\DotNetMapReduceWordCount\DotNetMapReduceWordCount.sln

把hadoop用docker compose啟動

使用指令把hadoop啟動：docker-compose up -d。 會看到：

1. 執行完有1個master 2個worker啟動
2. 在YARN的web節點看到有兩個Node
3. 在DataNode看到有兩個節點

把.Net core程式複製到master的hadoop節點裡面

把剛剛發佈出來的.Net core程式複製到master裡面，並且進入到master裡面的bash並且可以看到有copy進去的內容

docker cp dotnetmapreduce hadoop-dotnet-master:/dotnetmapreduce
docker exec -it hadoop-dotnet-master bash
ls
ls /dotnetmapreduce

把要計算的檔案放到hadoop的HDFS

透過下面指令把檔案放到hadoop的HDFS的input資料夾並且檢查：

hadoop fs -mkdir -p /input
hadoop fs -copyFromLocal /dotnetmapreduce/jane_austen.txt /input
hadoop fs -ls /input
                        

用hadoop Streaming執行net core mapreduce

用hadoop的streaming執行：

hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.7.2.jar \
    -files "/dotnetmapreduce" \
    -mapper "dotnet dotnetmapreduce/DotNetMapReduceWordCount.Mapper.dll" \
    -reducer  "dotnet dotnetmapreduce/DotNetMapReduceWordCount.Reducer.dll" \
    -input /input/* -output /output
                        

檢查結果

執行完了之後，可以看到計算的每個字出現次數

hadoop fs -ls /output
hadoop fs -cat /output/part-00000

會注意到這邊的結果和java版本有點不同，因為判斷字的邏輯不同導致。

如果docker不需要了，可以用docker-compose down把整個hadoop關掉。