package mmd.hw;

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

import org.apache.hadoop.conf.*;
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.mapreduce.*;
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.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.*;


public class MatrixMultiplex
{
    Integer inputSize = new Integer(500);

    // Mapper
    public static class Matrix_Mapper extends Mapper<LongWritable, Text, Text, Text> {
        public void map(LongWritable key, Text value, Context mapOutput) throws IOException, InterruptedException {
            String line = value.toString();
            String[] splitLineInput = line.split(",");

            Text outputKey = new Text();
            Text outputValue = new Text();

            if (splitLineInput[0].equals("M")){
                for (int k=0; k<inputSize; k++){
                    outputKey.set(splitLineInput[1] + "," + k);
                    outputValue.set(splitLineInput[0] + "," + splitLineInput[2] + "," + splitLineInput[3]);
                    mapOutput.write(outputKey, outputValue);
                }
            } else{
                for (int i=0; i<inputSize; i++){
                    outputKey.set(i + "," + splitLineInput[2]);
                    outputValue.set(splitLineInput[0] + "," + splitLineInput[1] + "," + splitLineInput[3]);
                    mapOutput.write(outputKey, outputValue);
                }
            }
        }
    }

    // Reducer
    public static class Matrix_Reducer extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context combineOutput) throws IOException, InterruptedException {
            String[] value, nMatch;

            ArrayList<String> arrayM = new ArrayList<String>();
            ArrayList<String> arrayN = new ArrayList<String>();

            Integer product = new Integer(0);

            for (Text val : values) {
                value = val.toString().split(",");

                if (value[0].equals("M")) {
                    arrayM.add(value[1]);
                    arrayM.add(value[2]);
                } else {
                    arrayN.add(value[1]);
                    arrayN.add(value[2]);
                }
            }

            for (int i=0; i<arrayM.size(); i+=2) {
                for (int j=0; j<arrayN.size(); j+=2) {
                    if (arrayM.get(i).equals(arrayN.get(j))) {
                        product += Integer.parseInt(arrayM.get(i+1))*Integer.parseInt(arrayN.get(j+1));
                    }
                }
            }

            combineOutput.write(key, new Text(product.toString()));
        }
    }

    // main function
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

        if (otherArgs.length != 2) {
            System.err.println("Usage: matrix multiplication <in> <out>");
            System.exit(2);
        }

        Job job = new Job(conf, "matrix multi");
        job.setJarByClass(MatrixMultiplex.class);

        job.setMapperClass(Matrix_Mapper.class);
        job.setReducerClass(Matrix_Reducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

@@ -726,24 +726,31 @@
 iplex%0A%7B%0A    
+static 
 Integer inpu
@@ -773,9 +773,9 @@
 ger(
-5
+1
 00);
@@ -2152,60 +2152,43 @@
     
-ArrayList%3CString%3E arrayM = new ArrayList%3CString%3E(
+Integer product = new Integer(0
 );%0A
+%0A
     
@@ -2199,60 +2199,42 @@
     
-ArrayList%3CString%3E
+int%5B%5D
  array
-N
+M
  = new 
-ArrayList%3CString%3E();%0A
+int%5BinputSize%5D;
 %0A   
@@ -2246,40 +2246,41 @@
     
-Integer product
+int%5B%5D arrayN
  = new 
-Integer(0)
+int%5BinputSize%5D
 ;%0A%0A 
@@ -2442,54 +2442,55 @@
 rayM
-.add(value%5B1%5D);%0A                    arrayM.add
+%5BInteger.parseInt(value%5B1%5D)%5D = Integer.parseInt
 (val
@@ -2552,54 +2552,55 @@
 rayN
-.add(value%5B1%5D);%0A                    arrayN.add
+%5BInteger.parseInt(value%5B1%5D)%5D = Integer.parseInt
 (val
@@ -2672,292 +2672,75 @@
 ; i%3C
-arrayM.size(); i+=2) %7B%0A                for (int j=0; j%3CarrayN.size(); j+=2) %7B%0A                    if (arrayM.get(i).equals(arrayN.get(j))) %7B%0A                        product += Integer.parseInt(arrayM.get(i+1))*Integer.parseInt(arrayN.get(j+1));%0A                    %7D%0A                %7D
+inputSize; i++) %7B%0A                product += arrayM%5Bi%5D * arrayN%5Bi%5D;
 %0A   
@@ -3052,24 +3052,95 @@
 ingArgs();%0A%0A
+        conf.set(%22mapreduce.output.textoutputformat.separator%22, %22,%22);%0A%0A
         if (
@@ -3232,14 +3232,30 @@
  %3Cin
-%3E %3Cout
+put file%3E %3Coutput file
 %3E%22);

update MatrixMultiplex.java