Hadoop - Как получить месяц с максимальным индексом гриппа среди набора данных Google Flu Trends?

Question

Я пытаюсь написать простую программу Map Reduce с использованием Hadoop, которая даст мне месяц, наиболее подверженный гриппу.Я использую набор данных трендов Google Flu, который можно найти здесь http://www.google.org/flutrends/data.txt.

Я написал и Mapper и редуктор, как показано ниже

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

private static final Log LOG =
        LogFactory.getLog(MaxFluPerMonthMapper.class);

@Override
protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
    String row = value.toString();
    LOG.debug("Received row " + row);
    List<String> columns = Arrays.asList(row.split(","));
    String date = columns.get(0);
    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
    int month = 0;
    try {
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(sdf.parse(date));
        month = calendar.get(Calendar.MONTH);
    } catch (ParseException e) {
        e.printStackTrace();
    }
    for (int i = 1; i < columns.size(); i++) {
        String fluIndex = columns.get(i);
        if (StringUtils.isNotBlank(fluIndex) && StringUtils.isNumeric(fluIndex)) {
            LOG.info("Writing key " + month + " and value " + fluIndex);
            context.write(new IntWritable(month), new IntWritable(Integer.valueOf(fluIndex)));
        }
    }
}

}

Редуктор

public class MaxFluPerMonthReducer extends Reducer<IntWritable, IntWritable, Text, IntWritable> {
private static final Log LOG =
        LogFactory.getLog(MaxFluPerMonthReducer.class);

@Override
protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
                    throws IOException, InterruptedException {
    LOG.info("Received key " + key.get());
    int sum = 0;
    for (IntWritable intWritable : values) {
        sum += intWritable.get();
    }
    int month = key.get();
    String monthString = new DateFormatSymbols().getMonths()[month];
    context.write(new Text(monthString), new IntWritable(sum));
}

}

С помощью этих картографов и редукторов, показанных выше, я получаю следующий вывод

январь 545419 февраль 528022 март 436348 апрель 336759 май 346482 июнь309795 июль 312966 август 307346 сентябрь 322359 октябрь 428346 ноябрь 461195 декабрь 480078

То, что я хочу, - это всего лишь один вывод, дающий мне января 545419 Как мне этого добиться?путем сохранения состояния в редукторе или есть какое-то другое решение?или мой маппер и редуктор неверны для вопроса, который я задаю в этом наборе данных?

Answer 1

Вы можете добавить еще один шаг MapReduce.Mapper примерно так:

public class MyMapper extends Mapper<LongWritable, Text, Text, Text> {

@Override
protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
    // emit only first row
    if (key == 0)    
    {
        String row = value.toString();
        String[] values = row.split("\t");
        context.write(new Text(values[0]), new Text(values[1]));
    }

    }
}

Редуктор должен испускать все свои входные данные (которые будут только одной записью) напрямую для вывода.Количество картографов и редукторов должно быть установлено равным одному.Если в вашем задании MapReduce используется более одного редуктора, вам следует использовать один промежуточный шаг MapReduce для объединения результатов в один файл после задания MapReduce.Но этот способ кажется не очень эффективным.

Answer 2

Проблема в том, что Редуктор не имеет представления о других клавишах (по замыслу). Можно было бы настроить другой редуктор, чтобы найти максимальное значение, учитывая все данные из вашего текущего редуктора. Однако это излишне, так как вы знаете, что у вас будет только 12 записей, которые вам нужно обработать, а настройка другого Reducer будет иметь больше затрат, чем просто запуск последовательного сценария.

Я бы предложил написать какой-нибудь другой скрипт для обработки вашего текстового вывода.