São fornecidas partes de uma
implementação do MapReduce que suporta operações distribuídas e não
distribuídas (apenas as partes chatas). Você buscará o software inicial com o git (um sistema de controle de versão). Para aprender
mais sobre o git, consulte o
manual do usuário do git.
O software fornecido foi
desenvolvido para trabalhar com o sistema operacional Unix.
A URL que vai ser usada é https://github.com/BU-CS451-DS/goals-2018.git. Para instalar os arquivos na sua
máquina, você precisa clonar o repositório, executando os comandos abaixo.
$ git clone https://github.com/BU-CS451-DS/goals-2018.git 451$ cd 451$ lsMakefile srcO Git permite que você acompanhe as
alterações feitas no código. Por exemplo, se você quiser verificar seu
progresso, poderá confirmar suas alterações executando:
$ git commit -am 'partial solution to lab 1'A implementação Map/Reduce fornecida
oferece suporte para dois modos de operação, sequencial e distribuído. No
primeiro, as tarefas de mapeamento e redução são todas executadas em série:
primeiro, a primeira tarefa de mapeamento é executada até a conclusão, depois a
segunda, depois a terceira, etc. Quando todas as tarefas de mapeamento são
concluídas, a primeira tarefa de redução é executada, depois a segunda etc.
Esse modo, embora não seja muito rápido, pode ser muito útil para depuração,
pois remove grande parte do ruído visto em uma execução paralela. O modo distribuído
executa muitas threads de trabalho que primeiro executam tarefas de mapeamento
em paralelo e, em seguida, reduzem as tarefas. Isso é muito mais rápido, mas
também mais difícil de implementar e depurar.
O pacote mapreduce fornece uma biblioteca
Map/Reduce simples com uma implementação sequencial. Os aplicativos
distribuídos normalmente devem chamar Distributed () [localizado em master.go]
para iniciar um trabalho, mas podem chamar Sequential () [também em master.go]
para obter uma execução sequencial para fins de depuração.
O fluxo da implementação do mapreduce é o
seguinte:
f0-0, ..., f0-0, f0-, ..., f<#files-1>-0, ... f<#files-1>-. Você terá que escrever/modificar doMap, doReduce, e schedule . Estes procedimentos estão localizados em common_map.go, common_reduce.go, e schedule.go respectivamente. Você também terá que escrever as
funções map e reduce em../main/wc.go.
A implementação Map / Reduce que você
clonou está com algumas partes faltando. Antes de poder escrever seu primeiro
par de funções Map/Reduce, será necessário corrigir a implementação sequencial.
Em particular, faltam duas partes cruciais no código fornecido a você: a função
que divide a saída de uma tarefa de mapeamento e a função que reúne todas as
entradas para uma tarefa reduzida.Estas tarefas são realizadas pela função doMap() em common_map.go, e a função doReduce() em common_reduce.go respectivamente. Observe os comentários
nestes dois arquivos.
Para auxiliar na verificação se suas
implementações de doMap() e de
doReduce() estão corretas, você deve usar o
conjunto de testes Go. Estes testes estão implementados no arquivo test_test.go. Para executar os testes para a
implementação sequencial que você consertou, execute:
$ export GOROOT=/research/sesa/451/go$ export PATH=$PATH:$GOROOT/bin$ export GOPATH=$HOME/451$ cd "$GOPATH/src/mapreduce"$ go test -run Sequential mapreduce/... ok mapreduce 2.694s
Você recebe crédito total por esta parte se o seu
software passar nos testes sequenciais (conforme executado pelo comando acima).
Se a saída não for OK, sua implementação
tem algum erro. Para tentar obter mais informação sobre a execução, coloque debugEnabled = true em common.go, e adicione -v para o comando de teste acima. Você obterá mais informações como abaixo:
$ go test -v -run Sequential mapreduce/...
=== RUN TestSequentialSinglemaster: Starting Map/Reduce task testMerge: read mrtmp.test-res-0master: Map/Reduce task completed--- PASS: TestSequentialSingle (1.34s)=== RUN TestSequentialManymaster: Starting Map/Reduce task testMerge: read mrtmp.test-res-0Merge: read mrtmp.test-res-1Merge: read mrtmp.test-res-2master: Map/Reduce task completed--- PASS: TestSequentialMany (1.33s)PASS
ok mapreduce 2.672s
Agora que as tarefas de mapeamento e
redução estão conectadas, você vai começar
a implementar algumas operações interessantes de Map/ Reduce. Neste trabalho,
você implementará a contagem de palavras
- um exemplo simples e clássico de Map/ Reduce. Especificamente, sua tarefa é
modificar mapF e reduceF de modo que wc.go reporte o número de ocorrências de cada palavra. Uma palavra é
qualquer sequência contígua de letras, como determinado em unicode.IsLetter.
Existem alguns arquivos de entrada pg-*.txt em ~/451/src/main, que foram obtidos de Project
Gutenberg. Tente compilar o software inicial clonado e o execute com os
arquivos de entrada fornecidos:
$ export GOROOT=/research/sesa/451/go$ export PATH=$PATH:$GOROOT/bin$ export GOPATH=$HOME/451$ cd "$GOPATH/src/main"$ go run wc.go master sequential pg-*.txt# command-line-arguments./wc.go:14: missing return at end of function./wc.go:21: missing return at end of functionA compilação falha porque as funções de
mapeamento (mapF()) e redução (reduceF()) em wc.go ainda não estão finalizadas. Antes que
você comece a codificá-las, leia a Seção 2 do artigo MapReduce. Suas
funções mapF() e reduceF() vão ser um pouco diferentes daquelas da
Seção 2.1 do artigo. Sua função mapF() recebe o nome do arquivo e seu conteúdo, deve separá-lo em palavras, e retornar uma
estrutura slice de pares key/value, do tipo mapreduce.KeyValue. Seu reduceF() será acionado uma vez para cada chave, com
um slice de todos os valores sendo gerado por mapF() para esta chave, e deve ser retornado um único valor.
Você pode testar a solução usando:
$ cd "$GOPATH/src/main"$ time go run wc.go master sequential pg-*.txtmaster: Starting Map/Reduce task wcseqMerge: read mrtmp.wcseq-res-0Merge: read mrtmp.wcseq-res-1Merge: read mrtmp.wcseq-res-2master: Map/Reduce task completed14.59user 3.78system 0:14.81elapsedA saída estará no arquivo "mrtmp.wcseq".
Você pode remover o arquivo de saída e todos os arquivos intermediários com:
$ rm mrtmp.*
Sua implementação estará correta se o comando
a seguir produzir as seguintes 10 palavras:
$ sort -n -k2 mrtmp.wcseq | tail -10
he: 34077was: 37044that: 37495I: 44502in: 46092a: 60558to: 74357of: 79727and: 93990the: 154024Para verificar o teste mais rapidamente,
execute:
$ sh ./test-wc.sh
que reportará se sua solução é correta ou
não.
Você recebe crédito total por esta parte se a saída
do map/ reduce de contagem de palavras corresponder à saída correta para a execução
sequencial acima.
·
Você pode encontrar um bom material de strings em Go em Go Blog on strings.
·
Você pode usar strings.FieldsFunc para dividir uma string em
componentes.
·
O pacote strconv é conveniente (http://golang.org/pkg/strconv/) para
converter strings para inteiros etc.
Um das
maiores vantagens do Map / Reduce é que o desenvolvedor não precisa estar
ciente de que seu código está sendo executado em paralelo em muitas máquinas.
Em teoria, poderemos pegar o código de contagem de palavras que você escreveu
acima e paralelizá-lo automaticamente!
A implementação atual executa todo o
mapeamento e reduz as tarefas uma após a outra no mestre. Embora isso seja
conceitualmente simples, não é ótimo para desempenho. Nesta parte do trabalho,
você implementará uma versão do MapReduce que divide o trabalho em um conjunto
de threads de trabalho, a fim de explorar vários núcleos. Embora o trabalho não
seja distribuído por várias máquinas, como nas implementações de Map/Reduce
"reais", sua implementação usará RPC e canais para simular uma
computação verdadeiramente distribuída.
Para coordenar a execução paralela das
tarefas, será usada uma thread mestre especial, que entrega o trabalho para os
trabalhadores e os espera finalizar o trabalho. O mestre só pode se comunicar
com os trabalhadores via RPC. São
fornecidos o código do trabalhador (mapreduce/worker.go), o código que inicializa os trabalhadores,
e o código que trata as mensagens RPC (mapreduce/common_rpc.go).
Seu trabalho é completar schedule.go no pacote mapreduce . Em
particular, você deve modificar schedule() in schedule.go
para entregar as tarefas de mapeamento e redução para os trabalhadores, e
retornar somente quando todas as tarefas forem finalizadas.
Examine run() em master.go. Ela chama sua função schedule() para executar as tarefas de mapeamento e redução, e chama merge() para reunir as saídas de cada tarefa de
redução em um único arquivo de saída. schedule só precisa informar aos trabalhadores o nome
do arquivo de entrada (mr.files[task]) e a tarefa task; cada trabalhador sabe de quais arquivos eles devem ler suas
entradas e em quais arquivos devem ser escritas as saídas. O mestre informa ao trabalhador
sobre a nova tarefa enviando uma chamada RPC Worker.DoTask, enviando o objeto DoTaskArgs com um argumento RPC.
Quando um trabalhador inicia, ele envia
um Register RPC para o mestre. mapreduce.go já implementa o manipulador do mestre de RPC Master.Register, e passa as informações do novo trabalhador para mr.registerChannel. Sua função schedule deve processar registros de novos trabalhadores
lendo deste canal. .
Informações sobre o job corrente sendo
executado estão na struct Master , definida em master.go. Observe que o mestre não precisa saber quais funções de
Mapeamento ou Redução estão sendo usadas para o trabalho; os trabalhadores
cuidarão da execução do código correto para Map ou Reduce (as funções corretas
são dadas a eles quando iniciadas por main/wc.go).
Para testar sua solução, use o mesmo
conjunto de teste usando na Parte I,trocando -run Sequential
por -run TestBasic. Isso executará o caso de teste
distribuído sem falhas de trabalho, em vez dos sequenciais que estavam sendo
executados antes:
$ go test -run TestBasic mapreduce/...Você recebe o ponto se sua implementação passar no TestBasic
de test_test.go .
·
O mestre deve enviar RPCs para os trabalhadores em paralelo de modo que os trabalhadores possam
trabalhar em suas tarefas de modo concorrente. Você irá achar a instrução go útil para este propósito
·
O mestre pode ter que esperar que um trabalhador finalize sua
tarefa antes de enviar mais tarefas para ele. Você pode achar canais úteis para
sincronizar threads que estão esperando por uma resposta com o mestre quando a
resposta chega. Canais são explicados no documento Concurrency
in Go.
·
A maneira mais fácil de depurar o programa é inserir
instruções debug(), colocar as saídas em um arquivo com go
test -run TestBasic mapreduce/... > out,
e pensar se a saída corresponde ao seu entendimento de como seu código deve se
comportar.
O código fornecido executa os trabalhadores como threads
em um único processo UNIX e pode explorar vários núcleos em uma única máquina.
Algumas modificações seriam necessárias para executar os trabalhadores em
várias máquinas que se comunicam através de uma rede. Os RPCs teriam que usar
TCP em vez de soquetes de domínio UNIX; uma maneira de iniciar processos de
trabalho em todas as máquinas deveria ser criada; e todas as máquinas teriam
que compartilhar armazenamento através de algum tipo de sistema de arquivos de
rede.
Nesta parte, você fará o mestre gerenciar
trabalhadores com falha. O MapReduce torna isso relativamente fácil porque os
trabalhadores não têm estado persistente. Se um trabalhador falhar, qualquer
RPC que o mestre emitiu para esse trabalhador falhará (por exemplo, devido a um
tempo limite). Portanto, se a chamada RPC do mestre ao trabalhador falhar, o
mestre deverá atribuir novamente a tarefa atribuída ao
trabalhador com falha a outro trabalhador.
Uma falha de RPC não significa
necessariamente que o trabalhador falhou; o trabalhador pode estar inacessível,
mas ainda executando. Assim, pode acontecer que dois trabalhadores recebam a
mesma tarefa e a executem. No entanto, como as tarefas são idempotentes, não
importa se a mesma tarefa é calculada duas vezes - nas duas vezes, ela gera a
mesma saída. Portanto, você não precisa fazer nada de especial neste caso. (Nos
testes, os trabalhadores nunca falham no meio da tarefa, portanto, você nem
precisa se preocupar com vários trabalhadores gravando no mesmo arquivo de
saída.)
Você não precisa lidar com falhas do mestre; pode
assumir que ele não falhará. Tornar o mestre tolerante a falhas é mais difícil
porque ele mantém um estado persistente que precisaria ser recuperado para
retomar as operações após uma falha principal.
Sua implementação tem que passar pelos
testes que estão faltando em test_test.go. O primeiro caso testa a falha de um
trabalhador, enquanto o segundo caso testa o gerenciamento de muitas falhas dos trabalhadores.
Periodicamente, os casos de teste iniciam novos trabalhadores que o mestre pode
usar para enviar novas tarefas, mas esses trabalhadores falham após manipular
algumas tarefas. Para executar esses testes:
$ go test -run Failure mapreduce/...Você recebe o ponto desta parte se seu software
passar os testes acima.
Você pode executar todos os testes executando
o script src/main/test-mr.sh. Com a solução correta, seu resultado
deve ser:
$ sh ./test-mr.sh==> Part Iok mapreduce 3.053s ==> Part IIPassed test ==> Part IIIok mapreduce 1.851s ==> Part IVok mapreduce 10.650s ==> Part V (challenge)Passed test
Antes de
enviar, execute todos os testes uma última vez. Você é responsável por garantir
que o seu código funcione.
$ sh ./test-mr.sh
Você deve enviar seu código via Classroom.