Hadoop（三）通過C#/python實現Hadoop MapReduce

MapReduce

Hadoop中將數據切分成塊存在HDFS不同的DataNode中，如果想彙總，按照常規想法就是，移動數據到統計程式：先把數據讀取到一個程式中，再進行彙總。

但是HDFS存的數據量非常大時，對彙總程式所在的伺服器將產生巨大壓力，並且網路IO也十分消耗資源。

為瞭解決這種問題，MapReduce提出一種想法：將統計程式移動到DataNode，每台DataNode（就近）統計完再彙總，充分利用DataNode的計算資源。YARN的調度決定了MapReduce程式所在的Node。

MapReduce過程

1. 確保數據存在HDFS上
2. MapReduce提交給ResourceManager（RM），RM創建一個Job。
3. 文件分片，預設將一個數據塊作為一個分片。
4. Job提交給RM，RM根據Node狀態選擇一臺合適的Node調度AM，AM向RM申請資源，RM調度合適的NM啟動Container，Container執行Task。
5. Map的輸出放入環形記憶體緩衝區，緩存溢出時，寫入磁碟，寫入磁碟有以下步驟
   1. 預設根據Hash分區，分區數取決於Reduce Task的數，相同Key的記錄被送到相同Reduce處理
   2. 將Map輸出的結果排序
   3. 將Map數據合併
   4. MapTask處理後產生多個溢出文件，會將多個溢出文件合併，生成一個經過分區和排序的MapOutFile(MOF)，這個過程稱為Spill
6. MOF輸出到3%時開始進行Reduce Task
7. MapTask與ReduceTask之間傳輸數據的過程稱為Shuffle。

下麵這個圖描述了具體的流程

Hadoop Streaming

Hadoop中可以通過Java來編寫MapReduce，針對不熟悉Java的開發者，Hadoop提供了通過可執行程式或者腳本的方式創建MapReduce的Hadoop Streaming。

Hadoop streaming處理步驟

hadoop streaming通過用戶編寫的map函數中標準輸入讀取數據（一行一行地讀取），按照map函數的處理邏輯處理後，將處理後的數據由標準輸出進行輸出到下一個階段。

reduce函數也是按行讀取數據，按照函數的處理邏輯處理完數據後，將它們通過標準輸出寫到hdfs的指定目錄中。

不管使用的是何種編程語言，在map函數中，原始數據會被處理成<key,value>的形式，但是key與value之間必須通過\t分隔符分隔，分隔符左邊的是key，分隔符右邊的是value,如果沒有使用\t分隔符，那麼整行都會被當作key

C#版MapReduce

首先，新增測試數據

vi mpdata

I love Beijing
I love China
Beijing is the capital of China

然後，將文件上傳到hdfs

[root@localhost ~]# hadoop fs -put mrdata /chesterdata

新建dotnet6的console項目mapper，修改Program.cs

using System;
using System.Text.RegularExpressions;

namespace mapper
{
    class Program
    {
        static void Main(string[] args)
        {
            string line;
            //Hadoop passes data to the mapper on STDIN
            while((line = Console.ReadLine()) != null)
            {
                // We only want words, so strip out punctuation, numbers, etc.
                var onlyText = Regex.Replace(line, @"\.|;|:|,|[0-9]|'", "");
                // Split at whitespace.
                var words = Regex.Matches(onlyText, @"[\w]+");
                // Loop over the words
                foreach(var word in words)
                {
                    //Emit tab-delimited key/value pairs.
                    //In this case, a word and a count of 1.
                    Console.WriteLine("{0}\t1",word);
                }
            }
        }
    }
}

發佈mapper

cd /demo/dotnet/mapper/
dotnet publish  -c Release -r linux-x64 /p:PublishSingleFile=true

新建dotnet6的console項目reducer，修改Program.cs

using System;
using System.Collections.Generic;

namespace reducer
{
    class Program
    {
        static void Main(string[] args)
        {
            //Dictionary for holding a count of words
            Dictionary<string, int> words = new Dictionary<string, int>();

            string line;
            //Read from STDIN
            while ((line = Console.ReadLine()) != null)
            {
                // Data from Hadoop is tab-delimited key/value pairs
                var sArr = line.Split('\t');
                // Get the word
                string word = sArr[0];
                // Get the count
                int count = Convert.ToInt32(sArr[1]);

                //Do we already have a count for the word?
                if(words.ContainsKey(word))
                {
                    //If so, increment the count
                    words[word] += count;
                } else
                {
                    //Add the key to the collection
                    words.Add(word, count);
                }
            }
            //Finally, emit each word and count
            foreach (var word in words)
            {
                //Emit tab-delimited key/value pairs.
                //In this case, a word and a count of 1.
                Console.WriteLine("{0}\t{1}", word.Key, word.Value);
            }
        }
    }
}

發佈reducer

/demo/dotnet/reducer
dotnet publish  -c Release -r linux-x64 /p:PublishSingleFile=true

執行mapepr reduce

hadoop jar /usr/local/hadoop323/hadoop-3.2.3/share/hadoop/tools/lib/hadoop-streaming-3.2.3.jar -input /chesterdata/mrdata -output /dotnetmroutput -mapper "./mapper" -reducer "./reducer" -file /demo/dotnet/mapper/bin/Release/net6.0/linux-x64/publish/mapper -f /demo/dotnet/reducer/bin/Release/net6.0/linux-x64/publish/reducer

查看mapreduce結果

[root@localhost reducer]# hadoop fs -ls /dotnetmroutput

-rw-r--r--   1 root supergroup          0 2022-05-01 16:40 /dotnetmroutput/_SUCCESS
-rw-r--r--   1 root supergroup         55 2022-05-01 16:40 /dotnetmroutput/part-00000

查看part-00000內容

[root@localhost reducer]# hadoop fs -cat /dotnetmroutput/part-00000

Beijing 2
China   2
I       2
capital 1
is      1
love    2
of      1
the     1

可以看到dotnet模式的Hadoop Streaming已經執行成功。

Python版MapReduce

# mapper.py
import sys
import re
p = re.compile(r'\w+')
for line in sys.stdin:
    words = line.strip().split(' ')
    for word in words:
        w = p.findall(word)
        if len(w) < 1:
            continue
        s = w[0].strip().lower()
        if s != "":
            print("%s\t%s" % (s, 1))

編寫reducer

# reducer.py
import sys
res = dict()
for word_one in sys.stdin:
    word, one = word_one.strip().split('\t')
    if word in res.keys():
        res[word] = res[word] + 1
    else:
        res[word] = 1
print(res)

執行mapreduce

hadoop jar /usr/local/hadoop323/hadoop-3.2.3/share/hadoop/tools/lib/hadoop-streaming-3.2.3.jar -input /chesterdata/mrdata -output /mroutput -mapper "python3 mapper.py" -reducer "python3 reducer.py" -file /root/mapper.py -file /root/reducer.py

查看mapreduce結果

[root@localhost lib]# hadoop fs -ls /mroutput

-rw-r--r--   1 root supergroup          0 2022-05-01 05:00 /mroutput/_SUCCESS
-rw-r--r--   1 root supergroup         89 2022-05-01 05:00 /mroutput/part-00000

查看part-00000內容

[root@localhost lib]# hadoop fs -cat /mroutput/part-00000

{'beijing': 2, 'capital': 1, 'china': 2, 'i': 2, 'is': 1, 'love': 2, 'of': 1, 'the': 1}

可以看到python模式的Hadoop Streaming已經執行成功。