MapReduce
Hadoop中将資料切分成塊存在HDFS不同的DataNode中,如果想彙總,按照正常想法就是,移動資料到統計程式:先把資料讀取到一個程式中,再進行彙總。
但是HDFS存的資料量非常大時,對彙總程式所在的伺服器将産生巨大壓力,并且網絡IO也十分消耗資源。
為了解決這種問題,MapReduce提出一種想法:将統計程式移動到DataNode,每台DataNode(就近)統計完再彙總,充分利用DataNode的計算資源。YARN的排程決定了MapReduce程式所在的Node。
MapReduce過程
- 確定資料存在HDFS上
- MapReduce送出給ResourceManager(RM),RM建立一個Job。
- 檔案分片,預設将一個資料塊作為一個分片。
- Job送出給RM,RM根據Node狀态選擇一台合适的Node排程AM,AM向RM申請資源,RM排程合适的NM啟動Container,Container執行Task。
- Map的輸出放入環形記憶體緩沖區,緩存溢出時,寫入磁盤,寫入磁盤有以下步驟
- 預設根據Hash分區,分區數取決于Reduce Task的數,相同Key的記錄被送到相同Reduce處理
- 将Map輸出的結果排序
- 将Map資料合并
- MapTask處理後産生多個溢出檔案,會将多個溢出檔案合并,生成一個經過分區和排序的MapOutFile(MOF),這個過程稱為Spill
- MOF輸出到3%時開始進行Reduce Task
- 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
使用與dotnet模式下同樣的測試資料,編寫mapper
# 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已經執行成功。