Paralelização Rubi redutores em Hadoop?

Question

Um redutor simples wordcount em Ruby parece com isso:

#!/usr/bin/env ruby
wordcount = Hash.new
STDIN.each_line do |line|
keyval = line.split("|")
wordcount[keyval[0]] = wordcount[keyval[0]].to_i+keyval[1].to_i
end

wordcount.each_pair do |word,count|
puts "#{word}|#{count}"
end

fica na STDIN todos os mapeadores valores intermediários. Não de uma chave específica. Então, na verdade há apenas um redutor para todos (e não redutor por palavra ou por conjunto de palavras).

No entanto, em exemplos Java Eu vi essa interface que recebe uma chave e uma lista de valores como inout. O que significa que os valores do mapa intermidiate são agrupados por chave antes reduzida e redutores podem ser executados em paralelo:

public static class Reduce extends MapReduceBase implements Reducer<Text, IntWritable, Text, IntWritable> {
            public void reduce(Text key, Iterator<IntWritable> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
              int sum = 0;
              while (values.hasNext()) {
                sum += values.next().get();
              }
              output.collect(key, new IntWritable(sum));
            }
          }

Esta é a única característica Java? Ou posso fazê-lo com Hadoop Transmissão usando Ruby?

Answer 1

Redutores sempre será executado em paralelo, se você estiver usando streaming de ou não (se você não está vendo isso, verifique se a configuração de trabalho está definido para permitir que múltiplos reduzir as tarefas - ver mapred.reduce.tasks no seu cluster ou configuração do trabalho). A diferença é que os pacotes de quadro as coisas um pouco mais bem para você quando você usa Java contra o fluxo.

Para Java, a reduzir a tarefa fica um iterador sobre todos os valores para uma determinada chave. Isto torna mais fácil para andar os valores se você estiver, digamos, somando a saída mapa na sua tarefa reduzir. Em streaming, você literalmente apenas obter um fluxo de pares chave-valor. Você são garantiu que os valores serão ordenados por chave, e que para uma determinada chave não vai ser dividida entre reduzir as tarefas, mas qualquer estado de rastreamento que você precisa é até você. Por exemplo, em Java sua saída mapa chega ao seu redutor simbolicamente na forma

key1, {val1, val2, val3} key2, {val7, Val8}

Com o streaming, a sua saída em vez parece

key1, val1 key1, val2 key1, val3 key2, val7 key2, Val8

Por exemplo, para escrever um redutor que calcula a soma dos valores para cada chave, você vai precisar de uma variável para armazenar a última chave que você viu e uma variável para armazenar a soma. Cada vez que você ler um novo par de valores-chave, você faça o seguinte:

1. Verifique se a chave é diferente do que a última chave.
2. Se assim for, a saída de sua chave e soma atual e redefinir a soma de zero.
3. adicione o valor atual para a sua soma e último conjunto de chave para a chave atual.

HTH.

Answer 2

Eu não tentei Hadoop Streaming de mim, mas a partir da leitura dos docs eu acho que você pode conseguir um comportamento paralelo similar.

Em vez de passar uma chave com os valores associados a cada redutor, streaming grupo vontade a saída mapeador por chaves. Ele também garante que os valores com as mesmas chaves não será dividida em várias redutores. Isso é um pouco diferente da funcionalidade normal Hadoop, mas, mesmo assim, a reduzir o trabalho será distribuído ao longo de vários redutores.

Tente usar a opção -verbose para obter mais informações sobre o que está realmente acontecendo. Você também pode tentar experimentar com a opção -D mapred.reduce.tasks=X onde X é o número desejado de redutores.