使用 Hadoop MapReduce 从 CSV 文件中收集特定数据

Question

我需要一些有关 MapReduce 程序的帮助。我有一个包含 15 列的 CSV 文件。我正在尝试根据第三列的值（年份）从两列（市场和资助金额）中提取数据。

截至目前，我的程序为每个条目输出两列（市场和资助金额）的数据。我希望它输出的是指定年份每个市场的资助总额或指定年份范围内每个市场的资助总额。

我将在下面发布我的映射器代码以及示例数据条目。任何帮助将不胜感激！

public class FundingMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
private Text market = new Text();
private Text amount = new Text();

public void map(LongWritable key, Text value, OutputCollector<Text,Text> output, Reporter reporter) throws IOException {
    String line = value.toString();
    CSVReader R = new CSVReader(new StringReader(line));

    String[] ParsedLine = R.readNext();
    R.close();

    amount.set(ParsedLine[15]);
    market.set(ParsedLine[3]);

    output.collect(market, amount);
}
}

/organization/hashoff, #HASHOFF, |Digital Media|Internet|Social Media|, Digital Media, USA, CO, Denver, Denver, /funding-round/669d6203c0374e6cf0e8d10f75ba0b8a, debt_financing, 12/8/14, 2014-12, 2014-Q4, 2014, 455,000

对于上述条目，我的程序将分别输出带有正确标题的 Digital Media 和 455,000 用于 Market 和 Amount Funded。我希望程序根据年份或指定的年份范围输出结果。

这也是我的工作代码：

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

    conf.setJobName("Funding Data");

    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(Text.class);
    conf.setMapOutputKeyClass(Text.class);
    conf.setMapOutputValueClass(Text.class);

    conf.setMapperClass(FundingMapper.class);
    conf.setNumReduceTasks(0);

    FileInputFormat.addInputPath(conf, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf, new Path(args[1]));
    JobClient.runJob(conf);
}
}

Answer 1

您将输出作为市场的关键和价值的数量，我认为您应该转向关于关键的年度市场连接。通过拥有这样一个键，您将能够生成一个 (key,value) 对的列表，如下例所示：

2014-DigitalMedia,455000
2014-OtherMarket,34500
2014-DigitalMedia,100000
2015-DigitalMedia,120000
2015-DigitalMedia,67000
2015-OtherMarket,15000
2015-OtherMarket,10000
...

然后，reducer 类可以获取每个元组并聚合每个键的数量，结果：

2014-DigitalMedia,555000
2014-OtherMarket,34500
2015-DigitalMedia,187000
2015-OtherMarket,25000

reducer 的代码可能是这样的：

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

在主程序中你必须添加：

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