hadoop矩阵乘法

引言

何为大矩阵？Excel、SPSS，甚至SAS处理不了或者处理起来非常困难，需要设计巧妙的分布式方法才能高效解决基本运算（如转置、加法、乘法、求逆）的矩阵，我们认为其可被称为大矩阵。这意味着此种矩阵的维度至少是百万级的、经常是千万级的、有时是亿万级的。举个形象的栗子。至2012年12月底，新浪微博注册用户数超5亿，日活跃用户4629万[1]，如果我们要探索这4000多万用户可以分成哪些类别，以便深入了解用户共同特征，制定精准营销策略，势必要用到聚类相关的算法（比如新浪大牛张俊林就利用聚类算法来挖掘新浪微博中的兴趣圈子[2]），而聚类算法都需要构造用户两两之间的关系，形成n*n的矩阵，称为相似度矩阵。新浪微博这个例子中，这个矩阵的维度是4000万*4000万。

大矩阵乘法为何重要？这个时代（我就不说那个被媒体用烂了的恶心词汇了），在海量数据中淘金，已是各大互联网公司的既定目标，亚马逊是数据化运营的成功典范，Google、百度投巨资用于对海量数据进行深度学习（Deep Learning）研究，阿里把数据与平台、金融并列成为未来三大战略。话扯得有点大而远，但任何伟大的战略，最终都要落地到非常细粒度的具体操作上。我们想在海量数据中淘到金子，强大的挖掘工具是必不可少的，而诸如回归、聚类、主成分分析、决策树等数据挖掘算法常常涉及大规模矩阵运算。这其中，大矩阵乘法具有较大的时间消耗，是算法的瓶颈。张俊林的文章[2]用到了谱聚类算法，其中有个重要步骤是将相似度矩阵转换为拉普拉斯矩阵，这就需要用到大矩阵乘法。很酷有没有！大矩阵乘法运算可以从根基上影响数据战略的实施，它比那些大而空的废话重要千百万倍。

我们将使用MapReduce来分布式计算大矩阵乘法。伟大导师黑格尔告诉我们，量变导致质变。当所操作的矩阵维度达到百万、千万级时，会产生亟待攻克的新问题：

  • 大矩阵如何存储？
  • 计算模型如何设计？
  • 矩阵维度如何传递给乘法运算？

第3个问题看似与矩阵的“大”无关，但实际上，当矩阵规模巨大时，我们就不太可能像对待小规模矩阵一样将整个矩阵读入内存、从而在一个job中就判断出其维度，而是需要分开成为两个job，第一个job专注于计算矩阵维度并存入全局变量，传递给第二个job做乘法运算。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

注意到，，这样的值不会在文件中存储。

计算模型

回顾一下矩阵乘法。

设，，那么

矩阵乘法要求左矩阵的列数与右矩阵的行数相等，的矩阵，与的矩阵相乘，结果为的矩阵。

现在我们来分析一下，哪些操作是相互独立的（从而可以进行分布式计算）。很显然，的计算和的计算是互不干扰的；事实上，中各个元素的计算都是相互独立的。这样，我们在Map阶段，可以把计算所需要的元素都集中到同一个key中，然后，在Reduce阶段就可以从中解析出各个元素来计算；的其他元素的计算同理。

我们还需要注意，会被、……的计算所使用，会被、……的计算所使用。也就是说，在Map阶段，当我们从HDFS取出一行记录时，如果该记录是的元素，则需要存储成个<key, value>对，并且这个key互不相同；如果该记录是的元素，则需要存储成个<key, value>对，同样的，个key也应互不相同；但同时，用于计算的、存放、……和、……的<key, value>对的key应该都是相同的，这样才能被传递到同一个Reduce中。

经过以上分析，整个计算过程设计为：

（1）在Map阶段，把来自表的元素，标识成条<key, value>的形式。其中，；把来自表的元素，标识成条<key, value>形式，其中，。

于是乎，在Map阶段，我们实现了这样的战术目的：通过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)解析出来，来自的元素，单独放在一个数组中，来自的元素，放在另一个数组中，然后，我们计算两个数组（各自看成一个向量）的点积，即可算出的值。

示例矩阵和相乘的计算过程如下图所示：

代码

主要代码如下：

[java] view plaincopyprint?

  1. import java.io.IOException;
  2. import java.util.HashMap;
  3. import java.util.Iterator;
  4. import java.util.StringTokenizer;
  5. import org.apache.hadoop.fs.FileSystem;
  6. import org.apache.hadoop.io.LongWritable;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.io.Writable;
  9. import org.apache.hadoop.io.WritableComparable;
  10. import org.apache.hadoop.mapred.FileSplit;
  11. import org.apache.hadoop.mapred.JobConf;
  12. import org.apache.hadoop.mapred.MapReduceBase;
  13. import org.apache.hadoop.mapred.Mapper;
  14. import org.apache.hadoop.mapred.OutputCollector;
  15. import org.apache.hadoop.mapred.RecordWriter;
  16. import org.apache.hadoop.mapred.Reducer;
  17. import org.apache.hadoop.mapred.Reporter;
  18. import org.apache.hadoop.mapred.TextOutputFormat;
  19. import org.apache.hadoop.mapred.lib.MultipleOutputFormat;
  20. import org.apache.hadoop.util.Progressable;
  21. public class Bigmmult {
  22. public static final String CONTROL_I = "\u0009";
  23. public static final int MATRIX_I = 4;
  24. public static final int MATRIX_J = 3;
  25. public static final int MATRIX_K = 2;
  26. public static String makeKey(String[] tokens, String separator) {
  27. StringBuffer sb = new StringBuffer();
  28. boolean isFirst = true;
  29. for (String token : tokens) {
  30. if (isFirst)
  31. isFirst = false;
  32. else
  33. sb.append(separator);
  34. sb.append(token);
  35. }
  36. return sb.toString();
  37. }
  38. public static class MapClass extends MapReduceBase implements
  39. Mapper<LongWritable, Text, Text, Text> {
  40. public static HashMap<String , Double> features = new HashMap<String, Double>();
  41. public void configure(JobConf job) {
  42. super.configure(job);
  43. }
  44. public void map(LongWritable key, Text value, OutputCollector<Text, Text> output,
  45. Reporter reporter) throws IOException, ClassCastException {
  46. // 获取输入文件的全路径和名称
  47. String pathName = ((FileSplit)reporter.getInputSplit()).getPath().toString();
  48. if (pathName.contains("m_ys_lab_bigmmult_a")) {
  49. String line = value.toString();
  50. if (line == null || line.equals("")) return;
  51. String[] values = line.split(CONTROL_I);
  52. if (values.length < 3) return;
  53. String rowindex = values[0];
  54. String colindex = values[1];
  55. String elevalue = values[2];
  56. for (int i = 1; i <= MATRIX_K; i ++) {
  57. output.collect(new Text(rowindex + CONTROL_I + i), new Text("a#"+colindex+"#"+elevalue));
  58. }
  59. }
  60. if (pathName.contains("m_ys_lab_bigmmult_b")) {
  61. String line = value.toString();
  62. if (line == null || line.equals("")) return;
  63. String[] values = line.split(CONTROL_I);
  64. if (values.length < 3) return;
  65. String rowindex = values[0];
  66. String colindex = values[1];
  67. String elevalue = values[2];
  68. for (int i = 1; i <= MATRIX_I; i ++) {
  69. output.collect(new Text(i + CONTROL_I + colindex), new Text("b#"+rowindex+"#"+elevalue));
  70. }
  71. }
  72. }
  73. }
  74. public static class Reduce extends MapReduceBase
  75. implements Reducer<Text, Text, Text, Text> {
  76. public void reduce(Text key, Iterator<Text> values,
  77. OutputCollector<Text, Text> output, Reporter reporter)
  78. throws IOException {
  79. int[] valA = new int[MATRIX_J];
  80. int[] valB = new int[MATRIX_J];
  81. int i;
  82. for (i = 0; i < MATRIX_J; i ++) {
  83. valA[i] = 0;
  84. valB[i] = 0;
  85. }
  86. while (values.hasNext()) {
  87. String value = values.next().toString();
  88. if (value.startsWith("a#")) {
  89. StringTokenizer token = new StringTokenizer(value, "#");
  90. String[] temp = new String[3];
  91. int k = 0;
  92. while(token.hasMoreTokens()) {
  93. temp[k] = token.nextToken();
  94. k++;
  95. }
  96. valA[Integer.parseInt(temp[1])-1] = Integer.parseInt(temp[2]);
  97. } else if (value.startsWith("b#")) {
  98. StringTokenizer token = new StringTokenizer(value, "#");
  99. String[] temp = new String[3];
  100. int k = 0;
  101. while(token.hasMoreTokens()) {
  102. temp[k] = token.nextToken();
  103. k++;
  104. }
  105. valB[Integer.parseInt(temp[1])-1] = Integer.parseInt(temp[2]);
  106. }
  107. }
  108. int result = 0;
  109. for (i = 0; i < MATRIX_J; i ++) {
  110. result += valA[i] * valB[i];
  111. }
  112. output.collect(key, new Text(Integer.toString(result)));
  113. }
  114. }
  115. }
时间： 2024-08-26 04:13:16

hadoop矩阵乘法的相关文章

矩阵乘法的MapReduce实现

对于任意矩阵M和N,若矩阵M的列数等于矩阵N的行数,则记M和N的乘积为P=M*N,其中mik 记做矩阵M的第i行和第k列,nkj记做矩阵N的第k行和第j列,则矩阵P中,第i行第j列的元素可表示为公式(1-1): pij=(M*N)ij=∑miknkj=mi1*n1j+mi2*n2j+--+mik*nkj (公式1-1) 由公式(1-1)可以看出,最后决定pij是(i,j),所以可以将其作为Reducer的输入key值.为了求出pij分别需要知道mik和nkj,对于mik,其所需要的属性有矩阵M,

矩阵乘法的Mapreduce实现过程

MapReduce实现矩阵的乘法 在学习pageRank算法时看到这么一个小小的编程应用.并且一直自诩只要有原理就能写出代码(只是时间问题),矩阵乘法的原理很简单,基本上理工科生(只要学过线性代数或者相关课程)都知道.但是从来没有想过通过并行计算的方式来完成矩阵乘法. 这里矩阵的知识就不啰嗦了,矩阵的乘积记作为P=M*N.则P中的元素 简单粗暴的讲,就是左矩阵M的行依次与右矩阵的列元素对应相乘,然后再相加. 可能说到mapreduce算法可能都会想到map用来整理数据(这里指的就是矩阵中的元素)

矩阵乘法的Strassen算法详解

题目描述 请编程实现矩阵乘法,并考虑当矩阵规模较大时的优化方法. 思路分析 根据wikipedia上的介绍:两个矩阵的乘法仅当第一个矩阵B的列数和另一个矩阵A的行数相等时才能定义.如A是m×n矩阵和B是n×p矩阵,它们的乘积AB是一个m×p矩阵,它的一个元素其中 1 ≤ i ≤ m, 1 ≤ j ≤ p. 值得一提的是,矩阵乘法满足结合律和分配率,但并不满足交换律,如下图所示的这个例子,两个矩阵交换相乘后,结果变了: 下面咱们来具体解决这个矩阵相乘的问题. 解法一.暴力解法 其实,通过前面的分析

51nod 1137 矩阵乘法

基本的矩阵乘法 中间for(int j=0;i<n;i++)  //这里写错了   应该是j<n 晚上果然  效率不行 等会早点儿睡 //矩阵乘法 就是 两个矩阵 第一个矩阵的列 等与 第二个矩阵的行相同 // 然后ans[i][j] += a[i][k] * b[k][j]; #include<bits/stdc++.h> using namespace std; typedef long long ll; const int maxn = 150; int n; ll a[ma

矩阵乘法

矩阵加法就是相同位置的数字加一下,矩阵减法也类似 矩阵乘以一个常数,就是所有位置都乘以这个数 矩阵乘以矩阵 计算规则是,第一个矩阵第一行的每个数字(2和1),各自乘以第二个矩阵第一列对应位置的数字(1和1),然后将乘积相加( 2 x 1 + 1 x 1),得到结果矩阵左上角的那个值3 矩阵的本质就是线性方程式,两者是一一对应关系.如果从线性方程式的角度,理解矩阵乘法就毫无难度.下面是一组线性方程式 矩阵的最初目的,只是为线性方程组提供一个简写形式 下面是严格的证明.有三组未知数 x.y 和 t,

矩阵乘法&lt;简单总结&gt;

原理:矩阵相乘最重要的方法是一般矩阵乘积.它只有在第一个矩阵的 行数 和第二个矩阵的 列数 相同时才可进行.若A为m×n矩阵,B为n×p矩阵,则他们的乘积AB会是一个m×p矩阵. 若A=    a    b    c        d    e    f        g    h    i    B=    A    D        B    E        C    F    A*B=CC=    aA+bB+cC    aD+bE+cF        dA+eB+fC    dD+eE

POJ2778 DNA Sequence Trie+矩阵乘法

题意:给定N个有A C G T组成的字符串,求长度为L的仅由A C G T组成的字符串中有多少个是不含给定的N个字符串的题解: 首先我们把所有的模式串(给定的DNA序列)建Trie,假定我们有一个匹配串,并且在匹配过程到S[i]这个字符时匹配到了Trie上的某个节点t,那么有两种可能: 匹配失败:t->child[S[i]]为空,跳转到t->fail,因此t->fail一定不能是某个模式串的结尾: 匹配成功:跳转到t->child[S[i+1]],因此t->child[S[i

【CDQ】BZOJ 2738 矩阵乘法

题意:给你一个N*N的矩阵,不用算矩阵乘法,但是每次询问一个子矩形的第K小数 思路: 整体二分+二维树状数组 二分询问的答案mid,将数值小等mid的全部插入二维树状数组 然后查询每个矩阵内的元素个数,若数量>K-1则放左边,否则放右边 继续向下分治,左边二分l-mid,右边mid-r 代码: #include<iostream> #include<cstdio> #include<cstring> #include<cstdlib> #include

codevs矩阵乘法系列

T1:矩阵乘法板子题,练手. #include <map> #include <set> #include <cmath> #include <ctime> #include <queue> #include <stack> #include <cstdio> #include <string> #include <vector> #include <cstdlib> #include