Странная проблема с форматированием в программе «карта / уменьшенное изображение» в Java

Question

У меня есть CSV-файл со следующими примерами записей.

| publisher  | site               | ad clicks | ad views |
|============|====================|===========|==========|
| publisher1 | www.sampleSite.com |        50 |       75 |
| publisher1 | www.sampleSite2.com|        10 |       40 |
| publisher2 | www.newSite1.com   |       100 |      175 |
| publisher2 | www.newSite2.com   |        50 |       65 |

Используя карту / уменьшение в Java, я пытаюсь суммировать все клики и просмотры объявлений для каждого издателя.Выходные данные должны быть такими:

publisher1 60, 115
publisher2 150, 240

Я написал следующий код.

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class SGSClickViewStats
{
    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> 
    {
        int recNo = 0;
        private Text publisherName = new Text();
        private Text mapOpValue    = new Text();

        public void map(LongWritable key, Text inputs, OutputCollector <Text, Text> output, Reporter rptr) throws IOException{
            String line = inputs.toString();
            String [] fields = line.split(",");
            String pubName = formatStats(fields[0]);
            String click   = fields[2];
            String views   = fields[3];
            // ***** send stats to reducer as a string separated by :
            String value   = click+":"+views;

            mapOpValue.set(formatStats(value));
            publisherName.set(pubName);   

            output.collect(publisherName, mapOpValue);
        }

        private String formatStats(String stat) {
            while((stat.indexOf("\"") >= 0) && (stat.indexOf(",")) >= 0){
                stat = stat.replace("\"","");
                stat = stat.replace(",","");
            }
            return stat;
        }
    }

    public static class Reduce extends MapReduceBase implements Reducer< Text, Text, Text, Text >
    {
        private Text pubName = new Text();
        public void reduce(Text key, Iterator<Text> value, OutputCollector<Text, Text> oc, Reporter rptr) throws IOException {
            int views     = 0;
            int clicks    = 0;
            String val    = "";
            String opVal  = "";
            Text textOpVal= new Text();

            while(value.hasNext()){
                val = value.next().toString();

                String [] tokens = val.split(":");

                try {
                    clicks = clicks + Integer.parseInt(tokens[0]);
                    views  = views  + Integer.parseInt(tokens[1]);
                } catch (Exception e) {
                    System.out.println("This is Command HQ, code red\nError Message: "+e.getLocalizedMessage()+" Error class: "+e.getClass()+"Extra, Array length: "+tokens.length);
                }
            }           

            try {
                            // ******* want to separate stats by comma but can't !!
                opVal = Integer.toString(clicks) + ":"+ Integer.toString(views);
            } catch (Exception e) {
                System.out.println("This is Command HQ, code Yellow\nError Message: "+e.getLocalizedMessage()+" Error class: "+e.getClass());
            }
            textOpVal.set(opVal);
            oc.collect(key, textOpVal);     
        }
    }

    public static void main(String [] args) throws Exception {
        JobConf jc = new JobConf(SGSClickViewStats.class);
        jc.setJobName("SGSClickViewStats");
        jc.setOutputKeyClass(Text.class);
        jc.setOutputValueClass(Text.class);

        jc.setMapperClass(Map.class);
        jc.setReducerClass(Reduce.class);
        jc.setCombinerClass(Reduce.class);

        jc.setInputFormat(TextInputFormat.class);
        jc.setOutputFormat(TextOutputFormat.class);

        FileInputFormat.setInputPaths(jc, new Path(args[0]));
        FileOutputFormat.setOutputPath(jc, new Path(args[1]));

        JobClient.runJob(jc);
    }
}

Эта программа работает нормально, но при выводе редуктора я не могу разделить окончательную статистикучерез запятую, которая является вторым комментарием с **.Если я это сделаю, все мои статистические данные станут равными 0, 0. Я получаю эту ошибку, когда пытаюсь отделить запятую.

Error Message: For input string: "50, 75" 
Error class: class java.lang.NumberFormatExceptionExtra, Array length: 1

Длина массива - это длина массива токенов в функции редуктора, поскольку я посылаю выводпреобразователь в разделитель двоеточий (:) разделенных токенов должен иметь 2 элемента, я вижу один, когда я устанавливаю вывод разделителя через запятую.

Я ссылался на множество статей, но не смог найти ответ.Я искренне надеюсь, что кто-то поможет!:)

Answer 1

Почему вы используете ваш редуктор в качестве комбайнера? К тому времени, когда вы переходите в фазу сокращения, это уже формат "издатель \ клики, просмотры" Я думаю, это может вызвать проблемы.

Можете ли вы прокомментировать следующую строку и проверить?

jc.setCombinerClass(Reduce.class);

Answer 2

NumberFormatException определенно отклоняется на Integer.parseInt, поэтому ваша ошибка должна быть с первой попытки, когда вы вычисляете сумму кликов и просмотров.Проверьте выходные данные, передаваемые картографом.Я почти уверен, что вы неправильно форматируете строки в маппере.

Редактировать : чтобы было ясно для будущих читателей: проблема заключалась в использовании класса Reducer также какОбъединение по ошибке, что приводит к выходу, отличному от ожидаемого на этапе карты.