Hadoop MapReduce Добавление значений с одинаковыми ключами - Java

Question

Я пытаюсь добавить номера для каждого повторяющегося имени. Однако я разделяю имена и цифры, но не могу понять, как добавить цифры. Пожалуйста, дайте мне знать, если вам нужна дополнительная информация, чтобы помочь.

Заранее спасибо.

---

Это мой код:

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

Answer 1

По какой-то причине вы используете редуктор по умолчанию в Hadoop, который является identityReducer. Я думаю, что это происходит потому, что в вашей функции reduce есть опечатка, поэтому вызывается функция сокращения класса ее родителя вместо той, которую вы создали. Чтобы избежать этой проблемы, всегда разумно использовать @Override в Java. Попробуйте написать @Override в вашей функции reduce и перекомпилировать.

Согласно исходному коду Hadoop , редуктор по умолчанию выглядит следующим образом:

/** 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, который в основном выполняет операцию сокращения, которую вы хотите выполнить, поэтому в вашей основной функции напишите:

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

Расположение пакета: org.apache.hadoop.mapred.lib.LongSumReducer