public static class MMMultTaskMapper extends
            Mapper<Object, Text, IntWritable, Text> {


        @Override
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {

            String name = ((FileSplit) context.getInputSplit()).getPath().getName();
            String[]  values = value.toString().split(" ");
            if(name.charAt(0)=='M'){
            //  System.out.println(values[1]);
                context.write(new IntWritable(Integer.parseInt(values[1])), new Text(("M " + values[0] + " " + values[2])));
            }else{
                context.write(new IntWritable(Integer.parseInt(values[0])), new Text(("N " + values[1] + " " + values[2])));
            }

        }
    }


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

        HashMap<String, Integer> nMap = new HashMap<String, Integer> ();
        HashMap<String, Integer> mMap = new HashMap<String, Integer> ();

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

            for(Text text:values){
                //System.out.println(text.toString());
                String[] singleValues = text.toString().split(" ");

                if(singleValues[0].equals("M")){
                    mMap.put(singleValues[1], Integer.parseInt(singleValues[2]));
                }else{
                    nMap.put(singleValues[1], Integer.parseInt(singleValues[2]));
                }
            }

            StringBuilder sb = new StringBuilder();

            for(String keyM : mMap.keySet()){
                for(String keyN : nMap.keySet()){
                    String result = keyM + " " + keyN + " " + Integer.toString(mMap.get(keyM) * nMap.get(keyN))+",";
                    //System.out.println(result);
                    sb.append(result);
                }
            }

            //System.out.println(list.toString());

            context.write(new Text(sb.toString()), key);

        }
    }

    public static class MMMultTask2Mapper extends
            Mapper<Object, Text, Text, IntWritable> {

        @Override
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            //System.out.println(value.toString());
                String[] editedList = value.toString().split(",");
                //System.out.println(editedList[1]);
                    for(int i = 0; i<editedList.length; i++){
                        //System.out.println(editedList[i]);

                        String[] line = editedList[i].split(" ");
                        //System.out.println(Integer.toString(line.length));
                        if(line.length>2){

                            String I = line[0];
                            String K = line[1];
                            String V = line[2];
                            context.write(new Text(I+" "+K), new IntWritable(Integer.parseInt(V)));

                        }
                    }
        }
    }

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

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

            int sum = 0;
            for(IntWritable value : values){
                sum += value.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }