import java.io.IOException;import java.util.*;import java.util.AbstractMap.SimpleEntry;import java.util.Map.Entry; import org.apache.hadoop.fs.Path;import org.apache.hadoop.conf.*;import org.apache.hadoop.io.*;import org.apache.hadoop.mapreduce.*;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; public class TwoStepMatrixMultiplication { public static class Map extends Mapper<LongWritable, Text, Text, Text> { public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString(); String[] indicesAndValue = line.split(","); Text outputKey = new Text(); Text outputValue = new Text(); if (indicesAndValue[0].equals("A")) { outputKey.set(indicesAndValue[2]); outputValue.set("A," + indicesAndValue[1] + "," + indicesAndValue[3]); context.write(outputKey, outputValue); } else { outputKey.set(indicesAndValue[1]); outputValue.set("B," + indicesAndValue[2] + "," + indicesAndValue[3]); context.write(outputKey, outputValue); } } } public static class Reduce extends Reducer<Text, Text, Text, Text> { public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException { String[] value; ArrayList<Entry<Integer, Float>> listA = new ArrayList<Entry<Integer, Float>>(); ArrayList<Entry<Integer, Float>> listB = new ArrayList<Entry<Integer, Float>>(); for (Text val : values) { value = val.toString().split(","); if (value[0].equals("A")) { listA.add(new SimpleEntry<Integer, Float>(Integer.parseInt(value[1]), Float.parseFloat(value[2]))); } else { listB.add(new SimpleEntry<Integer, Float>(Integer.parseInt(value[1]), Float.parseFloat(value[2]))); } } String i; float a_ij; String k; float b_jk; Text outputValue = new Text(); for (Entry<Integer, Float> a : listA) { i = Integer.toString(a.getKey()); a_ij = a.getValue(); for (Entry<Integer, Float> b : listB) { k = Integer.toString(b.getKey()); b_jk = b.getValue(); outputValue.set(i + "," + k + "," + Float.toString(a_ij*b_jk)); context.write(null, outputValue); } } } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = new Job(conf, "MatrixMatrixMultiplicationTwoSteps"); job.setJarByClass(TwoStepMatrixMultiplication.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setMapperClass(Map.class); job.setReducerClass(Reduce.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); job.waitForCompletion(true); }}
Notice that, in the first map-reduce step, there are Reduce jobs, each of which receives a column of elements from and a row of elements from and produces lines of output.
That is, in the first map-reduce step:
The number of Reduce job, which is equal to the number of unique keys that are sent to Reduce tasks, is .
The number of elements from that are sent to each Reduce job is .
The number of elements from that are sent to each Reduce job is .
The number of lines that are produced by each Reduce job is .
The communication cost of the first Map tasks is .
The communication cost of the first Reduce tasks is .
The communication cost of the second Map tasks is .
The communication cost of the second Reduce tasks is .
Like the One-Step Matrix Multiplication algorithm, if or is large, Reduce nodes may not have enough memory to store the entire column of and the entire row of . In the next post, we’ll see that we can partition the matrices into blocks to control memory requirements.
#Manpower Consultancy in Chennai , #Manpower Consultancy Chennai