Написание кода MApreduce для подсчета количества записей

Question

Я хочу написать код mapreduce для подсчета количества записей в данном файле CSV. Я не понимаю, что делать на карте и что делать в сокращении. Как мне решить эту проблему, может кто-нибудь что-то предложить?

Answer 1

• Ваша карта должна излучать 1 для каждой прочитанной записи
• ваш комбинатор должен излучать сумму всех полученных "1" (всего на карту)
• вы уменьшаетедолжно выделять общее количество записей

Answer 2

Ваш маппер должен испускать фиксированный ключ (просто используйте текст со значением «count») и фиксированное значение 1 (то же, что вы видите в примере с wordcount).

Тогда просто используйте LongSumReducer в качестве вашего редуктора.

Результатом вашей работы будет запись с ключом "count", а значением будет количество записей, которые вы ищете.

У вас есть возможность (кардинально!) Улучшить производительность, используя тот же LongSumReducer, что и комбайнер.

Answer 3

Надеюсь, у меня есть лучшее решение, чем принятый ответ.

Вместо того, чтобы выдавать 1 для каждой записи, почему бы нам не увеличить счетчик в map () и не увеличить счетчик после каждой задачи карты в cleanup ().

Промежуточные записи чтения могут быть уменьшены. А для редуктора нужно только агрегировать список из нескольких значений.

public class LineCntMapper extends
  Mapper<LongWritable, Text, Text, IntWritable> {

 Text keyEmit = new Text("Total Lines");
 IntWritable valEmit = new IntWritable();
 int partialSum = 0;

 public void map(LongWritable key, Text value, Context context) {
  partialSum++;
 }

 public void cleanup(Context context) {
  valEmit.set(partialSum);

   context.write(keyEmit, valEmit);

 }
}

Вы можете найти полный рабочий код здесь

Answer 4

Я бы просто использовал личность Mapper и личность Reducer.

Это Mapper.class и Reducer.class. Тогда просто прочитайте map input records

Вам действительно не нужно делать никакого кодирования, чтобы получить это.

Answer 5

import java.io.IOException;

import java.util.*;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.*;

import org.apache.hadoop.mapred.*;

public class LineCount 

{
    public static class Map extends MapReduceBase implements
            Mapper<LongWritable, Text, Text, IntWritable> 

{
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text("Total Lines");

    public void map(LongWritable key, Text value,
            OutputCollector<Text, IntWritable> output,Reporter reporter)
            throws IOException 
    {
        output.collect(word, one);
    }
}

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));
    }
}

public static void main(String[] args) throws Exception {
    JobConf conf = new JobConf(LineCount.class);

    conf.setJobName("LineCount");
    conf.setNumReduceTasks(5);
    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(IntWritable.class);

    conf.setMapperClass(Map.class);
    conf.setCombinerClass(Reduce.class);
    conf.setReducerClass(Reduce.class);

    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);

    FileInputFormat.setInputPaths(conf, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf, new Path(args[1]));

    JobClient.runJob(conf);
}
}

Answer 6

Используйте job.getcounters () , чтобы получить значения, которые вы увеличили для каждой записи после завершения задания.Если вы используете Java для написания задания mapreduce, тогда используйте enum для механизма подсчета.