前言
MapReduce打开了并行计算的大门,让我们个人开发者有了处理大数据的能力。但想用好MapReduce,把原来单机算法并行化,也不是一件容易事情。很多的时候,我们需要从单机算法能否矩阵化去思考,所以矩阵操作就变成了算法并行化的基础。
矩阵介绍
为了方便说明,举两个矩阵作为示例:
,
容易看出,是一个矩阵,是一个矩阵,我们能够算出:
这三个矩阵当然不大,但作为示例,它们将暂时享受大矩阵的待遇。
矩阵稀疏存储
理论上,在一个文件中存储4000万*4000万的矩阵当然是可以的,但非常失之优雅,因为这意味着在一条记录中挤下4000万个变量的值。
我们注意到,根据海量数据构造的矩阵,往往是极其稀疏的。比如4000万*4000万的相似度矩阵,一般来说,如果平均每个用户和1万个用户具有大于零的相似度,常识告诉我们,这样的关系网络已经非常密集了(实际网络不会这样密集,看看自己的微博,被你关注的、评论过的、转发过的对象,会达到1万个吗?);但对于4000万维度的矩阵,它却依然是极度稀疏的。
因此,我们可以采用稀疏矩阵的存储方式,只存储那些非零的数值。具体而言,存储矩阵的文件每一条记录的结构如下:
其中,第一个字段为行标签,第二个字段为列标签,第三个字段值为。
比如矩阵在HDFS中存储为
1 1 1 1 2 2 1 3 3 2 1 4 2 2 5 3 1 7 3 2 8 3 3 9 4 1 10 4 2 11 4 3 12
矩阵存储为
1 1 10 1 2 15 2 2 2 3 1 11 3 2 9
注意到,,这样的值不会在文件中存储。
MapReduce计算模型
回顾一下矩阵乘法。
设,,那么
矩阵乘法要求左矩阵的列数与右矩阵的行数相等,的矩阵,与的矩阵相乘,结果为的矩阵。
现在我们来分析一下,哪些操作是相互独立的(从而可以进行分布式计算)。很显然,的计算和的计算是互不干扰的;事实上,中各个元素的计算都是相互独立的。这样,我们在Map阶段,可以把计算所需要的元素都集中到同一个key中,然后,在Reduce阶段就可以从中解析出各个元素来计算;的其他元素的计算同理。
我们还需要注意,会被、……的计算所使用,会被、……的计算所使用。也就是说,在Map阶段,当我们从HDFS取出一行记录时,如果该记录是的元素,则需要存储成个<key,
value>对,并且这个key互不相同;如果该记录是的元素,则需要存储成个<key, value>对,同样的,个key也应互不相同;但同时,用于计算的、存放、……和、……的<key,
value>对的key应该都是相同的,这样才能被传递到同一个Reduce中。
经过以上分析,整个计算过程设计为:
1、在Map阶段,
把来自表的元素,标识成条<key, value>的形式。其中,;
把来自表的元素,标识成条<key, value>形式,其中,。
2、在Shuffle阶段,相同key的value会被加入到同一个列表中,形成<key, list(value)>对,传递给Reduce,这个由Hadoop自动完成。
3、在Reduce阶段,有两个问题需要自己问问:
- 当前的<key, list(value)>对是为了计算的哪个元素?
- list中的每个value是来自表或表的哪个位置?
第一个问题可以从key中获知,因为我们在Map阶段已经将key构造为形式。
第二个问题,也可以在value中直接读出,因为我们也在Map阶段做了标志。
接下来我们所要做的,就是把list(value)解析出来,来自的元素,单独放在一个数组中,来自的元素,放在另一个数组中,然后,我们计算两个数组(各自看成一个向量)的点积,即可算出的值。
示例矩阵和相乘的计算过程如下图所示:
测试数据
只存储非0的数据,3列存储,第一列“原矩阵行”,第二列“原矩阵列”,第三列“原矩阵值”。
sm1.csv
1,1,1 1,4,3 2,1,2 2,2,5 2,4,4 3,4,1 4,1,4 4,2,7 4,3,1 4,4,2<code> </code>
sm2.csv
1,1,5 2,2,2 4,1,3 4,2,1
程序代码
package org.conan.myhadoop.matrix; import java.io.IOException; import java.util.HashMap; import java.util.Iterator; import java.util.Map; 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.mapred.JobConf; 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.input.FileSplit; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.conan.myhadoop.hdfs.HdfsDAO; public class SparseMartrixMultiply { public static class SparseMatrixMapper extends Mapper>LongWritable, Text, Text, Text< { private String flag;// m1 or m2 private int rowNum = 4;// 矩阵A的行数 private int colNum = 2;// 矩阵B的列数 @Override protected void setup(Context context) throws IOException, InterruptedException { FileSplit split = (FileSplit) context.getInputSplit(); flag = split.getPath().getName();// 判断读的数据集 } @Override public void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException { String[] tokens = MainRun.DELIMITER.split(values.toString()); if (flag.equals("m1")) { String row = tokens[0]; String col = tokens[1]; String val = tokens[2]; for (int i = 1; i >= colNum; i++) { Text k = new Text(row + "," + i); Text v = new Text("A:" + col + "," + val); context.write(k, v); System.out.println(k.toString() + " " + v.toString()); } } else if (flag.equals("m2")) { String row = tokens[0]; String col = tokens[1]; String val = tokens[2]; for (int i = 1; i >= rowNum; i++) { Text k = new Text(i + "," + col); Text v = new Text("B:" + row + "," + val); context.write(k, v); System.out.println(k.toString() + " " + v.toString()); } } } } public static class SparseMatrixReducer extends Reducer>Text, Text, Text, IntWritable< { @Override public void reduce(Text key, Iterable>Text< values, Context context) throws IOException, InterruptedException { Map>String, String< mapA = new HashMap>String, String<(); Map>String, String< mapB = new HashMap>String, String<(); System.out.print(key.toString() + ":"); for (Text line : values) { String val = line.toString(); System.out.print("(" + val + ")"); if (val.startsWith("A:")) { String[] kv = MainRun.DELIMITER.split(val.substring(2)); mapA.put(kv[0], kv[1]); // System.out.println("A:" + kv[0] + "," + kv[1]); } else if (val.startsWith("B:")) { String[] kv = MainRun.DELIMITER.split(val.substring(2)); mapB.put(kv[0], kv[1]); // System.out.println("B:" + kv[0] + "," + kv[1]); } } int result = 0; Iterator>String< iter = mapA.keySet().iterator(); while (iter.hasNext()) { String mapk = iter.next(); String bVal = mapB.containsKey(mapk) ? mapB.get(mapk) : "0"; result += Integer.parseInt(mapA.get(mapk)) * Integer.parseInt(bVal); } context.write(key, new IntWritable(result)); System.out.println(); // System.out.println("C:" + key.toString() + "," + result); } } public static void run(Map>String, String< path) throws IOException, InterruptedException, ClassNotFoundException { JobConf conf = MainRun.config(); String input = path.get("input"); String input1 = path.get("input1"); String input2 = path.get("input2"); String output = path.get("output"); HdfsDAO hdfs = new HdfsDAO(MainRun.HDFS, conf); hdfs.rmr(input); hdfs.mkdirs(input); hdfs.copyFile(path.get("m1"), input1); hdfs.copyFile(path.get("m2"), input2); Job job = new Job(conf); job.setJarByClass(MartrixMultiply.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setMapperClass(SparseMatrixMapper.class); job.setReducerClass(SparseMatrixReducer.class); job.setInputFormatClass(TextInputFormat.class); job.setOutputFormatClass(TextOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2));// 加载2个输入数据集 FileOutputFormat.setOutputPath(job, new Path(output)); job.waitForCompletion(true); } }
增加SparseMartrixMultiply的启动配置
public static void main(String[] args) { sparseMartrixMultiply(); } public static void sparseMartrixMultiply() { Map<String, String> path = new HashMap<String, String>(); path.put("m1", "logfile/matrix/sm1.csv");// 本地的数据文件 path.put("m2", "logfile/matrix/sm2.csv"); path.put("input", HDFS + "/user/hdfs/matrix");// HDFS的目录 path.put("input1", HDFS + "/user/hdfs/matrix/m1"); path.put("input2", HDFS + "/user/hdfs/matrix/m2"); path.put("output", HDFS + "/user/hdfs/matrix/output"); try { SparseMartrixMultiply.run(path);// 启动程序 } catch (Exception e) { e.printStackTrace(); } System.exit(0); }
运行结果
Delete: hdfs://192.168.1.210:9000/user/hdfs/matrix
Create: hdfs://192.168.1.210:9000/user/hdfs/matrix
copy from: logfile/matrix/sm1.csv to hdfs://192.168.1.210:9000/user/hdfs/matrix/m1
copy from: logfile/matrix/sm2.csv to hdfs://192.168.1.210:9000/user/hdfs/matrix/m2
2014-1-15 11:57:31 org.apache.hadoop.util.NativeCodeLoader >clinit<
警告: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2014-1-15 11:57:31 org.apache.hadoop.mapred.JobClient copyAndConfigureFiles
警告: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
2014-1-15 11:57:31 org.apache.hadoop.mapred.JobClient copyAndConfigureFiles
警告: No job jar file set. User classes may not be found. See JobConf(Class) or JobConf#setJar(String).
2014-1-15 11:57:31 org.apache.hadoop.mapreduce.lib.input.FileInputFormat listStatus
信息: Total input paths to process : 2
2014-1-15 11:57:31 org.apache.hadoop.io.compress.snappy.LoadSnappy >clinit<
警告: Snappy native library not loaded
2014-1-15 11:57:31 org.apache.hadoop.mapred.JobClient monitorAndPrintJob
信息: Running job: job_local_0001
2014-1-15 11:57:31 org.apache.hadoop.mapred.Task initialize
信息: Using ResourceCalculatorPlugin : null
2014-1-15 11:57:31 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: io.sort.mb = 100
2014-1-15 11:57:31 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: data buffer = 79691776/99614720
2014-1-15 11:57:31 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: record buffer = 262144/327680
1,1 A:1,1
1,2 A:1,1
1,1 A:4,3
1,2 A:4,3
2,1 A:1,2
2,2 A:1,2
2,1 A:2,5
2,2 A:2,5
2,1 A:4,4
2,2 A:4,4
3,1 A:4,1
3,2 A:4,1
4,1 A:1,4
4,2 A:1,4
4,1 A:2,7
4,2 A:2,7
4,1 A:3,1
4,2 A:3,1
4,1 A:4,2
4,2 A:4,2
2014-1-15 11:57:31 org.apache.hadoop.mapred.MapTask$MapOutputBuffer flush
信息: Starting flush of map output
2014-1-15 11:57:31 org.apache.hadoop.mapred.MapTask$MapOutputBuffer sortAndSpill
信息: Finished spill 0
2014-1-15 11:57:31 org.apache.hadoop.mapred.Task done
信息: Task:attempt_local_0001_m_000000_0 is done. And is in the process of commiting
2014-1-15 11:57:32 org.apache.hadoop.mapred.JobClient monitorAndPrintJob
信息: map 0% reduce 0%
2014-1-15 11:57:34 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息:
2014-1-15 11:57:34 org.apache.hadoop.mapred.Task sendDone
信息: Task ‘attempt_local_0001_m_000000_0‘ done.
2014-1-15 11:57:34 org.apache.hadoop.mapred.Task initialize
信息: Using ResourceCalculatorPlugin : null
2014-1-15 11:57:34 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: io.sort.mb = 100
2014-1-15 11:57:34 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: data buffer = 79691776/99614720
2014-1-15 11:57:34 org.apache.hadoop.mapred.MapTask$MapOutputBuffer >init<
信息: record buffer = 262144/327680
1,1 B:1,5
2,1 B:1,5
3,1 B:1,5
4,1 B:1,5
2014-1-15 11:57:34 org.apache.hadoop.mapred.MapTask$MapOutputBuffer flush
信息: Starting flush of map output
1,2 B:2,2
2,2 B:2,2
3,2 B:2,2
4,2 B:2,2
1,1 B:4,3
2,1 B:4,3
3,1 B:4,3
4,1 B:4,3
1,2 B:4,1
2,2 B:4,1
3,2 B:4,1
4,2 B:4,1
2014-1-15 11:57:34 org.apache.hadoop.mapred.MapTask$MapOutputBuffer sortAndSpill
信息: Finished spill 0
2014-1-15 11:57:34 org.apache.hadoop.mapred.Task done
信息: Task:attempt_local_0001_m_000001_0 is done. And is in the process of commiting
2014-1-15 11:57:35 org.apache.hadoop.mapred.JobClient monitorAndPrintJob
信息: map 100% reduce 0%
2014-1-15 11:57:37 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息:
2014-1-15 11:57:37 org.apache.hadoop.mapred.Task sendDone
信息: Task ‘attempt_local_0001_m_000001_0‘ done.
2014-1-15 11:57:37 org.apache.hadoop.mapred.Task initialize
信息: Using ResourceCalculatorPlugin : null
2014-1-15 11:57:37 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息:
2014-1-15 11:57:37 org.apache.hadoop.mapred.Merger$MergeQueue merge
信息: Merging 2 sorted segments
2014-1-15 11:57:37 org.apache.hadoop.mapred.Merger$MergeQueue merge
信息: Down to the last merge-pass, with 2 segments left of total size: 436 bytes
2014-1-15 11:57:37 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息:
1,1:(B:1,5)(B:4,3)(A:1,1)(A:4,3)
1,2:(A:1,1)(A:4,3)(B:2,2)(B:4,1)
2,1:(B:1,5)(B:4,3)(A:1,2)(A:2,5)(A:4,4)
2,2:(A:1,2)(A:2,5)(A:4,4)(B:4,1)(B:2,2)
3,1:(B:1,5)(B:4,3)(A:4,1)
3,2:(A:4,1)(B:2,2)(B:4,1)
4,1:(B:4,3)(B:1,5)(A:1,4)(A:2,7)(A:3,1)(A:4,2)
4,2:(A:1,4)(A:2,7)(A:3,1)(A:4,2)(B:2,2)(B:4,1)
2014-1-15 11:57:37 org.apache.hadoop.mapred.Task done
信息: Task:attempt_local_0001_r_000000_0 is done. And is in the process of commiting
2014-1-15 11:57:37 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息:
2014-1-15 11:57:37 org.apache.hadoop.mapred.Task commit
信息: Task attempt_local_0001_r_000000_0 is allowed to commit now
2014-1-15 11:57:37 org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter commitTask
信息: Saved output of task ‘attempt_local_0001_r_000000_0‘ to hdfs://192.168.1.210:9000/user/hdfs/matrix/output
2014-1-15 11:57:40 org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
信息: reduce < reduce
2014-1-15 11:57:40 org.apache.hadoop.mapred.Task sendDone
信息: Task ‘attempt_local_0001_r_000000_0‘ done.
2014-1-15 11:57:41 org.apache.hadoop.mapred.JobClient monitorAndPrintJob
信息: map 100% reduce 100%
2014-1-15 11:57:41 org.apache.hadoop.mapred.JobClient monitorAndPrintJob
信息: Job complete: job_local_0001
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Counters: 19
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: File Output Format Counters
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Bytes Written=53
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: FileSystemCounters
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: FILE_BYTES_READ=2503
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: HDFS_BYTES_READ=266
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: FILE_BYTES_WRITTEN=126274
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: HDFS_BYTES_WRITTEN=347
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: File Input Format Counters
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Bytes Read=98
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Map-Reduce Framework
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Map output materialized bytes=444
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Map input records=14
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Reduce shuffle bytes=0
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Spilled Records=72
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Map output bytes=360
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Total committed heap usage (bytes)=764215296
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: SPLIT_RAW_BYTES=220
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Combine input records=0
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Reduce input records=36
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Reduce input groups=8
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Combine output records=0
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Reduce output records=8
2014-1-15 11:57:41 org.apache.hadoop.mapred.Counters log
信息: Map output records=36
注:以上模型和代码分别来自:
计算模型参考:http://blog.csdn.net/xyilu/article/details/9066973
代码详见:http://blog.fens.me/hadoop-mapreduce-matrix/