时间戳

首页 > 解决方案 > Hadoop MapReduce 使用相同的键添加值 - Java

java - Hadoop MapReduce 使用相同的键添加值 - Java

问题描述

我正在尝试为每个重复名称添加数字。但是，我将名称和数字分开，但我不知道如何添加数字。如果您需要更多信息来提供帮助，请告诉我。

先感谢您。

到目前为止，这是我的代码：

package hadoop.names;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.File;
import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class names_app {

    public static class GroupMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

        /** The name. */
        Text nameText = new Text();

        /** The count text. */
        IntWritable count = new IntWritable();


        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            String line = value.toString();
            String[] keyvalue = line.split(",");
            nameText.set(new Text(keyvalue[3]));
            count.set(Integer.parseInt(keyvalue[4]));
            context.write(nameText, count);

        }
    }


    public static class GroupReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

        public void reduce(Text key, Iterator<IntWritable> values, Context context) throws IOException,
                InterruptedException {

            int n = 0;
            while (values.hasNext()) {
                n = n + values.next().get();
            }
            context.write(key, new IntWritable(n));

        }

    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

        FileUtils.deleteDirectory(new File("/output/names"));
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "GroupMR");
        job.setJarByClass(names_app.class);
        job.setMapperClass(GroupMapper.class);
        job.setReducerClass(GroupReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.setMaxInputSplitSize(job, 10);
        FileInputFormat.setMinInputSplitSize(job, 100);
        FileInputFormat.addInputPath(job, new Path("/input_data/Sample_of_names.csv"));
        FileOutputFormat.setOutputPath(job, new Path("/output/names"));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

输入样本：

NJ,F,1910,Mary,593
NJ,F,1910,Helen,438
NJ,F,1910,Anna,355
NJ,F,1910,Margaret,311
NJ,F,1910,Elizabeth,260
NJ,F,1910,Dorothy,255
NJ,F,1910,Rose,201
NJ,F,1910,Ruth,188
NJ,F,1910,Mildred,174
NJ,F,1910,Florence,169
NJ,F,1910,Catherine,158
NJ,F,1910,Marie,152
NJ,F,1910,Lillian,130
NJ,F,1910,Alice,125
NJ,F,1910,Frances,124

原始数据集链接：https ://www.kaggle.com/datagov/usa-names

我得到以下输出为 csv：

Aaliyah,5
Aaron,14
Aaron,22
Aaron,11
Aaron,17
Aaron,24
Aaron,12
Aaron,241
Aaron,9
Aaron,11
Aaron,199
Aaron,16
Abbey,5
Abbie,5
Abbie,5
Abbie,5

我想拥有：

Aaliyah,5
Aaron,576
Abbey,5
Abbie,15

标签： javacsvhadoopmapreducereducers

解决方案

出于某种原因，您使用的是 Hadoop 中的默认减速器，即identityReducer. 我认为它正在发生，因为你的reduce函数中有一个错字，所以它的父类 reduce 函数被调用而不是你创建的那个。@Override为了避免这个问题，在 Java中使用总是明智的。尝试@Override在您的reduce函数中编写并重新编译。

根据Hadoop 的源代码，默认的 reducer 是这样的：

/** Writes all keys and values directly to output. */
public void reduce(K key, Iterator<V> values,
                   OutputCollector<K, V> output, Reporter reporter)
  throws IOException {
  while (values.hasNext()) {
    output.collect(key, values.next());
  }     
}

基本上它会溢出映射器的输出。

更好的解决方案

只需使用给定的Hadoop，LongSumReducer它基本上可以执行您想要执行的reduce操作，因此在您的主函数中编写：

job.setReducerClass(LongSumReducer<Text>.class);

包裹的位置在org.apache.hadoop.mapred.lib.LongSumReducer

推荐阅读