########################################## The implemention of the ReducerWorker module.# ReducerWorker use getRemoteFile() get its partitions of files from each mapper,# then mergeSortedFiles() merge them into an unreducedFile. In unreducedFile, pairs with the same key will near.# reduce() collect these (key , values) into collector, and write the collector into local file and upload it to DFS. # # @author Yang Fan(fyabc) # @author Guanji Gao(ggj)#########################################implement ReducerWorker; include "sys.m";include "lists.m";include "sort.m";include "tables.m";include "bufio.m"; include "reducerworker.m";include "mapreduce.m";include "mrutil.m";include "ioutil.m"; include "../dfs/dfsutil.m";include "../dfs/dfsclient.m"; sys : Sys;lists : Lists;sort : Sort;bufio : Bufio;tables : Tables; mrutil : MRUtil;ioutil : IOUtil;mapreduce : MapReduce;dfsutil : DFSUtil;dfsclient : DFSClient; FD : import sys;Iobuf : import bufio;ReducerTask : import mrutil;OutputCollector : import ioutil;DFSFile : import dfsutil;DFSChunk : import dfsutil;DFSNode : import dfsutil;DFSChunkCmp : import dfsutil;Strhash : import tables; reducerPath := "/appl/MR/task/";downloadMutex : chan of int; init(){ sys = load Sys Sys->PATH; lists = load Lists Lists->PATH; sort = load Sort Sort->PATH; bufio = load Bufio Bufio->PATH; mrutil = load MRUtil MRUtil->PATH; ioutil = load IOUtil IOUtil->PATH; tables = load Tables Tables->PATH; mrutil->init(); ioutil->init(); downloadMutex = chan [1] of int;} run(mapperFileAddrs : list of string , reducerTask : ref ReducerTask) : (int, string){ downloadMutex <-= 0; getmr(reducerTask); <-downloadMutex; folderName := reducerPath + "tasks_" + string reducerTask.id + "/"; sys->create(folderName, sys->OREAD, sys->DMDIR + 8r777); files : list of string; for(p := mapperFileAddrs; p != nil; p = tl p) { mapperFileAddr := hd p; if (ioutil->getRemoteFile(mapperFileAddr , mapperFileAddr + "_part_" + string reducerTask.partitionIndex , folderName) == nil) return (-1, mapperFileAddr); files = (folderName + mapperFileAddr + "_part_" + string reducerTask.partitionIndex) :: files; } unreducedFile := folderName + "unreducedFile"; sys->create(unreducedFile , sys->ORDWR , 8r755); ioutil->mergeSortedFiles(files , unreducedFile); collector := ref OutputCollector(nil); collector.collection = Strhash[ref IOUtil->KVs].new(10000, nil); reader := bufio->open(unreducedFile , bufio->OREAD); line , key , prevkey , value : string; values : list of string; prevkey = key = line = nil; while(1) { line = reader.gets('\n'); if (line == nil) { if(key != nil){ mapreduce->reduce(key , values , collector); } break; } if(line[len line - 1] == '\n') line = line[ : len line - 1]; (key , value) = ioutil->splitLine(line); if(prevkey != nil && key != prevkey) { mapreduce->reduce(prevkey , values , collector); values = nil; } values = value :: values; prevkey = key; } sys->create(folderName + string reducerTask.id + "_" + reducerTask.outputFile , sys->ORDWR , 8r755); buffer := bufio->open(folderName + string reducerTask.id + "_" + reducerTask.outputFile , bufio->OWRITE); output := collector.getMap(); for(i := 0; i < len output; i++ ) { kvs := output[i]; for ( ; kvs.values != nil ; kvs.values = tl (kvs.values)) { buffer.puts(kvs.key + " " + hd (kvs.values) + "\n"); } } buffer.flush(); buffer.close(); #ud(folderName , string reducerTask.id + "_" + reducerTask.outputFile , reducerTask.outputRep , reducerTask.outputSize); return (0, nil);} getmr(reducerTask : ref ReducerTask){ if(sys->open(reducerPath + reducerTask.mrClassName, Sys->OREAD) == nil) { dfsclient->init(); file := dfsclient->getFile(reducerTask.mrClassName); total := (lists->last(file.chunks)).offset + big (lists->last(file.chunks)).size; fd := sys->create(reducerPath + reducerTask.mrClassName, Sys->ORDWR, 8r600); dd(fd, file,big 0,total); } mapreduce = load MapReduce (reducerPath + reducerTask.mrClassName); mapreduce->init();} ud(folderName : string , fileName : string , replicas : int , chunkSize : int){ sys->chdir(folderName); if (dfsclient == nil) dfsclient = load DFSClient DFSClient->PATH; if (dfsutil == nil) { dfsutil = load DFSUtil DFSUtil->PATH; dfsutil->init(); } dfsclient->init(); dfsclient->createFile(fileName, replicas); fd := sys->open(fileName, Sys->ORDWR); if (fd == nil) exit; (nil, dir) := sys->fstat(fd); totalSize := dir.length; offset := big 0; while (totalSize > big chunkSize) { dfsclient->createChunk(fileName, offset, chunkSize); offset += big chunkSize; totalSize -= big chunkSize; } if (totalSize != big 0) dfsclient->createChunk(fileName, offset, int totalSize); file := dfsclient->getFile(fileName); for (p := file.chunks; p != nil; p = tl p) { chunk := hd p; dfsclient->writeChunk(chunk, chunk.offset, chunk.size, fd); } sys->print("%s", file.toString()); for (p = file.chunks; p != nil; p = tl p) { chunk := hd p; sys->print("\t%s", chunk.toString()); for (q := chunk.nodes; q != nil; q = tl q) sys->print("\t\t%s", (hd q).toString()); }} dd(fd : ref Sys->FD, file : ref DFSFile, offset : big, size : big){ if (dfsclient == nil) dfsclient = load DFSClient DFSClient->PATH; if (dfsutil == nil) { dfsutil = load DFSUtil DFSUtil->PATH; dfsutil->init(); } if (sys == nil) sys == load Sys Sys->PATH; sort = load Sort Sort->PATH; dfsclient->init(); chunks := file.getChunks(); cmp := ref DFSChunkCmp(); sort->sort(cmp, chunks); i := 0; while (i < len chunks && chunks[i].offset + big chunks[i].size < offset) i++; while (i < len chunks && chunks[i].offset < offset + size) { tempfd := dfsclient->readChunk(chunks[i]); start := big 0; buf := array [Sys->ATOMICIO] of byte; if (chunks[i].offset < offset) start = sys->seek(tempfd, offset - chunks[i].offset, Sys->SEEKSTART); if (chunks[i].offset + big chunks[i].size > offset + size) { total := int (offset + size - chunks[i].offset - start); while (total > len buf) { sys->read(tempfd, buf, len buf); sys->write(fd, buf, len buf); total -= len buf; } if (total > 0) { sys->read(tempfd, buf[:total], total); sys->write(fd, buf[:total], total); } } else while ((length := sys->read(tempfd, buf, len buf)) != 0) sys->write(fd, buf, length); sys->remove(sys->fd2path(tempfd)); i++; }}
########################################
#
# The implemention of the ReducerWorker module.
# ReducerWorker use getRemoteFile() get its partitions of files from each mapper,
# then mergeSortedFiles() merge them into an unreducedFile. In unreducedFile, pairs with the same key will near.
# reduce() collect these (key , values) into collector, and write the collector into local file and upload it to DFS.
#
# @author Yang Fan(fyabc)
# @author Guanji Gao(ggj)
#
########################################
implement ReducerWorker;
include "sys.m";
include "lists.m";
include "sort.m";
include "tables.m";
include "bufio.m";
include "reducerworker.m";
include "mapreduce.m";
include "mrutil.m";
include "ioutil.m";
include "../dfs/dfsutil.m";
include "../dfs/dfsclient.m";
sys : Sys;
lists : Lists;
sort : Sort;
bufio : Bufio;
tables : Tables;
mrutil : MRUtil;
ioutil : IOUtil;
mapreduce : MapReduce;
dfsutil : DFSUtil;
dfsclient : DFSClient;
FD : import sys;
Iobuf : import bufio;
ReducerTask : import mrutil;
OutputCollector : import ioutil;
DFSFile : import dfsutil;
DFSChunk : import dfsutil;
DFSNode : import dfsutil;
DFSChunkCmp : import dfsutil;
Strhash : import tables;
reducerPath := "/appl/MR/task/";
downloadMutex : chan of int;
init()
{
sys = load Sys Sys->PATH;
lists = load Lists Lists->PATH;
sort = load Sort Sort->PATH;
bufio = load Bufio Bufio->PATH;
mrutil = load MRUtil MRUtil->PATH;
ioutil = load IOUtil IOUtil->PATH;
tables = load Tables Tables->PATH;
mrutil->init();
ioutil->init();
downloadMutex = chan [1] of int;
}
run(mapperFileAddrs : list of string , reducerTask : ref ReducerTask) : (int, string)
{
downloadMutex <-= 0;
getmr(reducerTask);
<-downloadMutex;
folderName := reducerPath + "tasks_" + string reducerTask.id + "/";
sys->create(folderName, sys->OREAD, sys->DMDIR + 8r777);
files : list of string;
for(p := mapperFileAddrs; p != nil; p = tl p)
{
mapperFileAddr := hd p;
if (ioutil->getRemoteFile(mapperFileAddr , mapperFileAddr + "_part_" + string reducerTask.partitionIndex , folderName) == nil)
return (-1, mapperFileAddr);
files = (folderName + mapperFileAddr + "_part_" + string reducerTask.partitionIndex) :: files;
}
unreducedFile := folderName + "unreducedFile";
sys->create(unreducedFile , sys->ORDWR , 8r755);
ioutil->mergeSortedFiles(files , unreducedFile);
collector := ref OutputCollector(nil);
collector.collection = Strhash[ref IOUtil->KVs].new(10000, nil);
reader := bufio->open(unreducedFile , bufio->OREAD);
line , key , prevkey , value : string;
values : list of string;
prevkey = key = line = nil;
while(1)
{
line = reader.gets('\n');
if (line == nil)
{
if(key != nil){
mapreduce->reduce(key , values , collector);
}
break;
}
if(line[len line - 1] == '\n')
line = line[ : len line - 1];
(key , value) = ioutil->splitLine(line);
if(prevkey != nil && key != prevkey)
{
mapreduce->reduce(prevkey , values , collector);
values = nil;
}
values = value :: values;
prevkey = key;
}
sys->create(folderName + string reducerTask.id + "_" + reducerTask.outputFile , sys->ORDWR , 8r755);
buffer := bufio->open(folderName + string reducerTask.id + "_" + reducerTask.outputFile , bufio->OWRITE);
output := collector.getMap();
for(i := 0; i < len output; i++ )
{
kvs := output[i];
for ( ; kvs.values != nil ; kvs.values = tl (kvs.values))
{
buffer.puts(kvs.key + " " + hd (kvs.values) + "\n");
}
}
buffer.flush();
buffer.close();
#ud(folderName , string reducerTask.id + "_" + reducerTask.outputFile , reducerTask.outputRep , reducerTask.outputSize);
return (0, nil);
}
getmr(reducerTask : ref ReducerTask)
{
if(sys->open(reducerPath + reducerTask.mrClassName, Sys->OREAD) == nil)
{
dfsclient->init();
file := dfsclient->getFile(reducerTask.mrClassName);
total := (lists->last(file.chunks)).offset + big (lists->last(file.chunks)).size;
fd := sys->create(reducerPath + reducerTask.mrClassName, Sys->ORDWR, 8r600);
dd(fd, file,big 0,total);
}
mapreduce = load MapReduce (reducerPath + reducerTask.mrClassName);
mapreduce->init();
}
ud(folderName : string , fileName : string , replicas : int , chunkSize : int)
{
sys->chdir(folderName);
if (dfsclient == nil)
dfsclient = load DFSClient DFSClient->PATH;
if (dfsutil == nil) {
dfsutil = load DFSUtil DFSUtil->PATH;
dfsutil->init();
}
dfsclient->init();
dfsclient->createFile(fileName, replicas);
fd := sys->open(fileName, Sys->ORDWR);
if (fd == nil)
exit;
(nil, dir) := sys->fstat(fd);
totalSize := dir.length;
offset := big 0;
while (totalSize > big chunkSize) {
dfsclient->createChunk(fileName, offset, chunkSize);
offset += big chunkSize;
totalSize -= big chunkSize;
}
if (totalSize != big 0)
dfsclient->createChunk(fileName, offset, int totalSize);
file := dfsclient->getFile(fileName);
for (p := file.chunks; p != nil; p = tl p) {
chunk := hd p;
dfsclient->writeChunk(chunk, chunk.offset, chunk.size, fd);
}
sys->print("%s", file.toString());
for (p = file.chunks; p != nil; p = tl p) {
chunk := hd p;
sys->print("\t%s", chunk.toString());
for (q := chunk.nodes; q != nil; q = tl q)
sys->print("\t\t%s", (hd q).toString());
}
}
dd(fd : ref Sys->FD, file : ref DFSFile, offset : big, size : big)
{
if (dfsclient == nil)
dfsclient = load DFSClient DFSClient->PATH;
if (dfsutil == nil) {
dfsutil = load DFSUtil DFSUtil->PATH;
dfsutil->init();
}
if (sys == nil)
sys == load Sys Sys->PATH;
sort = load Sort Sort->PATH;
dfsclient->init();
chunks := file.getChunks();
cmp := ref DFSChunkCmp();
sort->sort(cmp, chunks);
i := 0;
while (i < len chunks && chunks[i].offset + big chunks[i].size < offset)
i++;
while (i < len chunks && chunks[i].offset < offset + size) {
tempfd := dfsclient->readChunk(chunks[i]);
start := big 0;
buf := array [Sys->ATOMICIO] of byte;
if (chunks[i].offset < offset)
start = sys->seek(tempfd, offset - chunks[i].offset, Sys->SEEKSTART);
if (chunks[i].offset + big chunks[i].size > offset + size) {
total := int (offset + size - chunks[i].offset - start);
while (total > len buf) {
sys->read(tempfd, buf, len buf);
sys->write(fd, buf, len buf);
total -= len buf;
}
if (total > 0) {
sys->read(tempfd, buf[:total], total);
sys->write(fd, buf[:total], total);
}
}
else while ((length := sys->read(tempfd, buf, len buf)) != 0)
sys->write(fd, buf, length);
sys->remove(sys->fd2path(tempfd));
i++;
}
}