Hadoop - 如何在谷歌流感趋势数据集中获得最大流感指数的月份？答案

【问题标题】：Hadoop - How to get the month with the maximum flu index across the Google Flu Trends dataset?Hadoop - 如何在谷歌流感趋势数据集中获得最大流感指数的月份？
【发布时间】：2011-10-17 16:59:18
【问题描述】：

我正在尝试使用 Hadoop 编写一个简单的 Map Reduce 程序，它将给出最容易患流感的月份。我正在使用谷歌流感趋势数据集，可以在这里找到http://www.google.org/flutrends/data.txt。

Mapper和reducer我都写了，如下图

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

}

使用上面显示的这些 Mapper 和 Reducer，我得到以下输出

一月 545419 二月 528022 三月 436348 4 月 336759 5 月 346482 六月 309795 七月 312966 八月 307346 九月 322359 十月 428346 十一月 461195 12 月 480078

我想要的只是一个输出给我 January 545419 我怎样才能做到这一点？通过将状态存储在减速器中还是有其他解决方案？或者我的 mapper 和 reducer 对于我在这个数据集上提出的问题是错误的？

【问题讨论】：

首先，使用组合器。其次，我认为实现这一点的一个 hacky 方法可能是使用 12 个 reducer。我不确定这是否总是有效。但是看看这个hadoop.apache.org/mapreduce/docs/r0.21.0/api/org/apache/hadoop/…

标签： java hadoop

【解决方案1】：

问题在于 Reducer 不知道其他键（按设计）。给定当前 reducer 的所有数据，可以设置另一个 Reducer 来查找最大值。但是，这太过分了，因为您知道您将只需要处理 12 条记录，并且设置另一个 Reducer 将比仅运行串行脚本有更多的开销。

我建议编写一些其他脚本来处理您的文本输出。

【讨论】：

【解决方案2】：

您可以再添加一个 MapReduce 步骤。映射器是这样的：

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

    }
}

Reducer 必须将其所有输入（将只有一条记录）直接发送到输出。映射器和减速器的数量应设置为 1。如果您的 MapReduce 作业使用多于一个 reducer，您应该在 MapReduce 作业之后使用另一个中间 MapReduce 步骤将结果合并到一个文件中。但这种方式似乎不是很有效。

【讨论】：