1 二阶范数 
 
 $$||A||^2 = tr(AA^T)  \quad\quad\quad\quad\quad\quad\quad\quad$$  

 2 矩阵求导基本公式（等式两边同时去除tr，公式不变）： 

 

 $$tr(A) = tr(A^T)
\quad\quad\quad\quad\quad\quad\quad\quad$$  

 

 $$\frac{\partial tr(AX)}{\partial X} = A^T  \quad\quad\quad\quad\quad\quad\quad\quad$$ 
 

 $$\frac{\partial tr(AX)}{\partial X^T} = A  
\quad\quad\quad\quad\quad\quad\quad\quad$$ 
 

 $$\frac{\partial tr(AXB)}{\partial X } = (BA)^T = A^TB^T  
\quad\quad\quad\quad\quad\quad\quad\quad$$  

 3 则简单公式均可以推导出来： 

 

 $$\frac{\partial tr(X^TAX)}{\partial X} = AX + (X^TA)^T = (A^T+A)X
\quad\quad\quad\quad\quad\quad\quad\quad$$  

 

 $$\frac{\partial tr(X^TAX)}{\partial X^T} = (AX)^T + X^TA = X^T(A^T+A)
	\quad\quad\quad\quad\quad\quad\quad\quad$$  

 

 $$\frac{\partial tr(ABA^TC)}{\partial A} = (BA^TC)^T+CAB=C^TAB^T+CAB
	\quad\quad\quad\quad\quad\quad\quad\quad$$ 

$\because \nabla f(x)=2Ax+2b$ 
 $\nabla f(x+\alpha d)=2A(x+\alpha d)+2b$ 
 又 $\because  \frac{ \partial h(\alpha)}{\partial (\alpha)} = \nabla f(x+\alpha d) \frac {\partial (x+\alpha d)}{\partial \alpha}   
  =\nabla f(x+\alpha d)d=0$ 
 $\therefore \alpha=-\frac {d^\intercal (2b+2Ax)}{2d^\intercal Ad}$ 
 $=-\frac {d^\intercal \nabla f(x)}{2d^\intercal Ad}$

 
基于二阶矩阵的优化问题（一）线搜索策略
 
非精确线搜索更新$X_{k+1}$
线搜索策略
最速下降+牛顿法
启发
    
修正牛顿法
特征值修正（eigenvalue modification）
谱分解
     
单位增量法
    
拟牛顿法
  
 

 
非精确线搜索更新$X_{k+1}$
 
优化算法的问题在于如何从$X_k$更新到$X_{k+1}$、确定步长公式详见 基于二阶矩阵的优化问题（二）和判定迭代终止的条件（详见 于二阶矩阵的优化问题（三））是我们需要在不精确搜索中解决的问题。
 我们有两种方法来解决这个问题：
 1.线搜索策略
 2.信任域策略
 
线搜索策略
 
我们选择一个下降方向来使得lost function慢慢迭代至最小值。
 
 
 
$X_{k+1}=X_k+{\alpha }_k{p}_k$
 
 
这是$X_{k+1}$的更新公式，可以看到，我们需要知道其${\alpha }_k$和$p_k$的值才能进行迭代（${\alpha }_k$代表了步长、$p_k$代表下降方向），第一种方法就是$p_k$走一个最速下降方向（负梯度），而${\alpha }_k$走一个极其猥琐的距离，但这种方法在函数比较诡异时效果较差（如rosenbrock函数），但在某些复杂问题时，我们还是使用最速下降来解决问题，详见 大规模的优化问题（一）。
 
最速下降+牛顿法
 
让我们来列一下牛顿方向的式子：
 
 
 
$f(x_k+p)\approx f_k+p^T\nabla f_k+\frac{1}{2}{p}^T{\nabla }^2{f}_{k}p$
 
 
在这里，我们要假设这里的二次型（${\nabla }^2{f}_k$）是正定的，这是函数的局部二次型构型就是碗状的，我们可以求出这里的pk（下降方向）：
 
 
 
$p_K^N=-({\nabla }^2{f}_k{)}^{-1}\nabla f_k$;
 
 
这边的$p_K^N$就是这里的局部二次型的顶点，在牛顿法中，我们可以一步就走到这个点。
 ok，这个问题的理论情况是这样的，但在实际问题中，我们不可能去求一个hessian矩阵的逆矩阵，所以，我们把$1/({\nabla }^2{f}_k)$移到等式的左边，变成：
 
 
 
$p_K^N\ast ({\nabla }^2{f}_k)=-\nabla f_k$，
 
 
这个等式实际上就等于Ax=b的形式，这边我们只需要求解一个线性方程组即可。
 以上为牛顿法的基础理论，下面看一下牛顿法的实现步骤：
 

 
 
可以看到，牛顿法的意义在于足够靠近真解时的快速收敛，我们先走几部最速下降法，使得足够靠近阈值，后再启动牛顿法，会有较好的效果,即保证了算法的效率，有保证了算法的鲁棒性。
 
下面是最速下降法+牛顿法matlab实现的代码：
 
function [x1] = min_hybrid(func, dfunc, d2func ,x, MAX_IT ,eps, threshold)
    err = 10.0;
    steps = 0;
    alpha = 1.0;
    //最速下降法
    while (err > threshold)//此处阈值需要调参
        f = func(x);
        g = dfunc(x);
        p = -g;
        step=step+1;
        alpha = backtracking(@func, x, f, g, p, 1e-4, 0.8, alpha * 2.0);
        x = x + alpha * p;
        if (steps > MAX_IT)
            break;
        end
        error = norm(g);
    end
    //牛顿法
    while (error > eps)
        f = func(x);
        g = dfunc(x);
        G = d2func(x);
        p = -G\g;
        x = x + p;
        step=step+1;
        if (steps > MAX_IT)
            break;
        end
        error = norm(g);
    end
	x1=x;
end

function alpha = backtracking(func, x, f, df, p, c, rho, alpha0)
    alpha = alpha0;
    while (func(x(1) + alpha * p(1), x(2) + alpha * p(2)) > f + c * alpha * df' * p)
        alpha = alpha * rho;
    end
end
 
其中func、gfunc和hess需要自行解析计算，func的接口只有x，gfunc和hess没有输入参数。（这边要注意，使用wolfe条件时，alpha可以恒取1）
 
启发
 
牛顿法实际上是在改进真解，用少量次数的迭代即可实现一阶方法上千次的迭代效果，首先系数矩阵必须要正定，否则牛顿法不能保证收敛。
 
修正牛顿法
 
牛顿法最大的问题在于，系数矩阵必须要正定，那么如果系数矩阵不正定，则将$B_k$改成正定（modification），下面是修改系数矩阵的算法框架：
 
 
 
初值$x_0$
 for k=0,1,2,…
         令 $B_k={\nabla }^2{f}_k+E_k$,其中$E_k$为修正项，使得$B_k$充分正定。
         求解 $B_k{p}_k=-\nabla f_k$
         $x_{k+1}=x_k+{\alpha }_k{p}_k$（其中${\alpha }_k$在wolfe条件时可以取1）
 end
 
 
好了，现在我们需要解决四件事，就可以实现修正牛顿法的运用：
 1.判定${\nabla }^2{f}_k$是否正定；
 2.解方程组$B_k{p}_k=-\nabla f_k$；
 3.如何构建$E_k$；
 4.如何使得$||E_k||$最小；
 
特征值修正（eigenvalue modification）
 
谱分解
 
首先我们先对对称矩阵$B_k$做正交分解（householder等），得到正交阵Q，和对角阵V，然后就可以得到$B_k$的所有特征值，我们知道，正定矩阵所有的特征值都是充分大的正数，即${\lambda }_{min}>\delta >>eps>0$，当出现负的特征值时，我们添加一个修正项$E_k$使得特征值大于等于δ。此时的$||E_k||$的Frobenius范数最小。
 即用Matlab中的eig()函数求出特征值后，直接对特征值修改即可。
 
单位增量法
 
谱分解的问题在于，其真正改变的是特征值，所以对$B_k$的改动是比较大的，当我们不希望$B_k$发生很大的变动时，我们就使用单位增量法。单位增量法直接对$B_k$添加一个修正项$E_{k}I$，使得其正定。
 首先，我们对$B_k$进行cholesky分解：
 
 
 
$B_k=LD{L}^T$
 
 
L是对角线均为1的下三角阵，D为对角阵，若$B_k$不定，D的对角元$d_{ii}$会过大。
 下面是单位增量法的Matlab代码：
 
function[L,D]=modification(A,delta,beta)
	if(norm(A-A')>eps
		return;
	end
	n=size(A,1);
	d=zero(n,1);
	L=zero(n);
	C=zero(n);
	theta=zeros(n,1);
	for j = 1:n
		C(j,j) = A(j,j)-sum(d(1:j-1)'.*L(j,1:j-1).^2);
		for i = j+1:n
			C(i,j) = A(i,j)-sum(d(1:j-1)'.*L(i,1:j-1).*L(j,1:j-1));
			absjj=abs(C(i,j));
			if theta < abs(C(i,j))
				theta = abs(C(i,j));
			end
		end
		d(j) = max([abs(C(j,j)), (theta/beta)^2, \delta]);
		for i = j+1:n
			L(i,j) = C(i,j)/d(j);
		end
		L(j,j) = 1.0;
	end
	D=diag(d);
end
 
单位增量法得到的$B_k$和$B_k+E_k$在形式上很接近，但其特征值完全不同，所以两种方法各有优势，可根据情况自行选择。
 
拟牛顿法
 
如果你的数据维数过大或者无法承受计算二阶矩阵的消耗，这边也可以使用拟牛顿法来计算下降方向，当然拟牛顿法也有其缺陷，下面我们来分析一下。
 首先我们介绍一下割线法，用弦的斜率近似代替目标函数的切线斜率，下面给出割线法的公式
 
 
 
$x_{k+1}=x_k-f(x_k)\ast \frac{x_{k-1}-x_k}{f(x_{k-1})-f(x_k)}$
 
 
这在本质上是计算函数的数值微分，但这种方法在接近收敛时会出现数值不稳定的情况，当接近收敛时，舍入误差占比变大，数值会发生很大的起伏变化。
 由此我们可以推导出拟牛顿法的公式
 
 
 
$x_{k+1}=x_k-\nabla f(x_k)\ast \frac{x_{k-1}-x_k}{\nabla f(x_{k-1})-\nabla f(x_k)}$
 
 
这里不在需要计算其hessian矩阵，由于hessian矩阵是$n^2$级别的，拟牛顿法对于大规模问题来说是非常节约资源的方法。
 
于是我们知道了
 
 
 
${\nabla }^{2}f(x_k)\ast (x_{k+1}-x_k)\approx \nabla f_{x+1}-\nabla f_k$
 
 
现在的情况与牛顿发不同，$\nabla f(x_k)$并不知道，这是一个不定方程组:
 
 
 
$B_{k+1}\ast s_k=y_k$
 
 
这里的$B_{k+1}$是$n^2$阶的矩阵，而我们只有n个方程，下面有SR1、BFGS等方法来求解$B_{k+1}$。
 详见 拟牛顿法的下降方向计算（一）
 此时，局部二次型的$p_K^N$方程变为
 
 
 
$p_K^N=-(B_k{)}^{-1}\nabla f_k$;
 
 
这里还有一条路，就是去构建$(B_k{)}^{-1}$,这时我们不需要再去求解线性方程组（这里设$H_k=(B_k{)}^{-1}$），这里$B_k$=I时，就是最速下降。
 
 
 
$p_k=-H_k\ast \nabla f_k$
 
 
这里有拟BFGS等方法来求解$H_k$,详见 拟牛顿法的下降方向计算（二）
 
github代码地址

二阶矩阵相乘公式
 

 
 
上例中的C11=A11*B11+A12*B21+A13*B31=1*3+0*2+2*1=5、C12=A11*B12+A12*B22+A13*B32=1*1+0*1+2*0=1
 

 
 
分析
 
 因为分布式计算的特点，需要找到相互独立的计算过程，以便能够在不同的节点上进行计算而不会彼此影响。根据矩
 
阵乘法的公式，C中各个元素的计算都是相互独立的，即各个cij在计算过程中彼此不影响。这样的话，在Map阶段可
 
以把计算所需要的元素都集中到同一个key中，然后，在Reduce阶段就可以从中解析出各个元素来计算cij。  另外，
 
以a11为例，它将会在c11、c12...c1p的计算中使用，以b11为例，它将会在c11、c21...cm1的计算中使用，也就是说，在Map阶段，当我们从HDFS取出一行记录时，如
 
果该记录是A的元素，则需要存储成p个<key, value>对，并且这p个key互不相同；如果该记录是B的元素，则需要存
 
储成m个<key, value>对，同样的，m个key也应互不相同；但同时，用于存放计算cij的ai1、ai2……ain和b1j、
 
b2j……bnj的<key, value>对的key应该都是相同的，这样才能被传递到同一个Reduce中。
 
 

 
 
设计
 
普遍有一个共识是：数据结构+算法=程序，所以在编写代码之前需要先理清数据存储结构和处理数据的算法。
 
 
Map阶段
 
 
在Map阶段，需要做的是进行数据准备。把来自矩阵A的元素aij，标识成p条<key, value>的形式，key="i,k",（其中
 
k=1,2,...,p），value="A,j,Aij"；把来自矩阵B的元素bij，标识成m条<key, value>形式，key="k,j"（其中
 
k=1,2,...,m），value="B,i,Bij"。  经过处理，用于计算cij需要的a、b就转变为有相同key（"i,j"）的数据对，通过value
 
中"A"、"B"能区分元素是来自矩阵A还是矩阵B，以及具体的位置（在矩阵A的第几列，在矩阵B的第几行）。
 
 
Shuffle阶段
 
这个阶段是Hadoop自动完成的阶段，具有相同key的value被分到同一个list中，形成<key,list(value)>对，再传递给Reduce。
 
 
Reduce阶段 
 
在Reduce阶段，有两个问题需要解决：
 
a. 当前的<key, list(value)>对是为了计算矩阵C的哪个元素？因为map阶段对数据的处理，key（i,j）中的数据对，就
 
是其在矩阵C中的位置，第i行j列。 
 
 
b. list中的每个value是来自矩阵A或矩阵B的哪个位置？这个也在map阶段进行了标记，对于value（x,y,z），只需要找
 
到y相同的来自不同矩阵（即x分别为A和B）的两个元素，取z相乘，然后加和即可。 
 
 
 
矩阵的两种表示方式
 
 
矩阵常用的两种表示方式，第一种是原始的表示方式，第二种是稀疏矩阵（只存储非0的元素）的表示方式。
 
第一种：使用最原始的表示方式，相同行内不同列数据通过","分割，不同行通过换行分割；  
 
 
第二种：通过行列表示法，即文件中的每行数据有三个元素通过分隔符分割，第一个元素表示行，第二个元素表示
 
列，第三个元素表示数据。这种方式对于可以不列出为0的元素，即可以减少稀疏矩阵的数据量。 
 

 
 

 
 

 
 
编写代码：
 
第一种数据结构
 
查看源数据：
 

[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/B
10,15
0,2
11,9
[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/A
1,2,3
4,5,0
7,8,9
10,11,12


 
 
MartrixMultiply：
 

package com.oner.mr.matrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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 com.oner.mr.util.HdfsDAO;

public class MartrixMultiply {

	public static class MatrixMapper extends
			Mapper<LongWritable, Text, Text, Text> {

		private String flag;// A or B;
		private int m = 4;// 矩阵A的行数
		private int p = 2;// 矩阵B的列数
		private int rowIndexA = 1; // 矩阵A，当前在第几行
		private int rowIndexB = 1; // 矩阵B，当前在第几行

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();
			flag = split.getPath().getName();// // 得到读取的矩阵名称
		}

		@Override
		protected void map(LongWritable k1, Text v1, Context context)
				throws IOException, InterruptedException {
			// 切分每行数据
			String[] fields = MainRun.DELIMITER.split(v1.toString());

			if (flag.equals("A")) {// 如果读的是矩阵A，则fields格式为{1,2,3},数组长度为3
				for (int k = 1; k <= p; k++) {// p表示矩阵B的列数
					Text key = new Text(rowIndexA + "," + k);//
					for (int j = 0; j < fields.length; j++) {// j代表矩阵A的当前列,fields.length表示矩阵A的列数，等于矩阵B的行数
						Text value = new Text("A," + (j + 1) + "," + fields[j]);// v的值为
						context.write(key, value);// 输出的数据格式key为(i,k),value为(A,j,Aij)。
						System.out.println(key.toString() + "  "
								+ value.toString());
					}

				}
				rowIndexA++; // 每执行一次map方法，矩阵向下移动一行

			} else if (flag.equals("B")) {// 如果读的是B,fields的格式为{10,15},数组长度为2
				for (int k = 1; k <= m; k++) {// m表示矩阵A的行数
					for (int j = 0; j < fields.length; j++) {// fields.length表示矩阵B的列数
						Text key = new Text(k + "," + (j + 1));
						Text value = new Text("B:" + rowIndexB + ","
								+ fields[j]);
						context.write(key, value);// 输出的数据格式key为(k,j),value为(B,i,Bij)。
						System.out.println(key.toString() + "  "
								+ value.toString());
					}
				}
				rowIndexB++;// 每执行一次map方法，矩阵向下移动一行
			}
		}

	}

	public static class MatrixReducer extends
			Reducer<Text, Text, Text, LongWritable> {
		@Override
		protected 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 value : values) {
				String val = value.toString();
				System.out.print("(" + val + ")");

				if (val.startsWith("A")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到A,j,Aij中的j,Aij
					mapA.put(kv[0], kv[1]);// 将j作为key,Aij作为value存入mapA

					// System.out.println("A:" + kv[0] + "," + kv[1]);

				} else if (val.startsWith("B")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到B,j,Bij中的i,Bij
					mapB.put(kv[0], kv[1]);// 将i作为key,Bij作为value存入mapB

					// System.out.println("B:" + kv[0] + "," + kv[1]);
				}
			}

			long result = 0;
			Iterator<String> mkeys = mapA.keySet().iterator();// 得到mapA所有的键集合
			while (mkeys.hasNext()) {
				String mkey = mkeys.next();
				if (mapB.get(mkey) == null) {// 因为mkey取的是mapA的key集合，所以只需要判断mapB是否存在即可。
					continue;
				}
				result += Long.parseLong(mapA.get(mkey))
						* Long.parseLong(mapB.get(mkey));
			}
			context.write(key, new LongWritable(result));
			System.out.println();

			// System.out.println("C:" + key.toString() + "," + result);

		}
	}

	public static void run(Map<String, String> path) throws IOException,
			ClassNotFoundException, InterruptedException {

		Configuration conf = new Configuration();

		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("A"), input1);
		hdfs.copyFile(path.get("B"), input2);

		Job job = Job.getInstance(conf);
		job.setJarByClass(MainRun.class);

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

		job.setMapperClass(MatrixMapper.class);
		job.setReducerClass(MatrixReducer.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);
	}
}


 
 
MainRun：
 

package com.oner.mr.matrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

/*
 * 驱动程序
 */
public class MainRun {

	public static final String HDFS = "hdfs://master:9000";
	public static final Pattern DELIMITER = Pattern.compile("[\t,]");

	public static void main(String[] args) throws ClassNotFoundException,
			IOException, InterruptedException {
		martrixMultiply();
	}

	private static void martrixMultiply() throws ClassNotFoundException,
			IOException, InterruptedException {
		Map<String, String> path = new HashMap<String, String>();
		path.put("A", "/home/hadoop/logfile/matrix/A.csv");// 本地的数据文件
		path.put("B", "/home/hadoop/logfile/matrix/B.csv");
		path.put("input", HDFS + "/user/hdfs/matrix");// HDFS的目录
		path.put("input1", HDFS + "/user/hdfs/matrix/A");
		path.put("input2", HDFS + "/user/hdfs/matrix/B");
		path.put("output", HDFS + "/user/hdfs/matrix/output");

		MartrixMultiply.run(path);
	}

}


 
 
打成jar包后运行：hadoop jar matrix.jar com.oner.mr.matrix.MainRun
 
查看结果：
 

[hadoop@master ~]$ hadoop fs -cat /user/hdfs/matrix/output/part-r-00000
1,1	43
1,2	46
2,1	40
2,2	70
3,1	169
3,2	202
4,1	232
4,2	280
绘图演示结果：
 

 
 

 
 
第二种数据结构
 
MainRun：
 

package com.oner.mr.sparsematrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

/*
 * 驱动程序
 */
public class MainRun {

	public static final String HDFS = "hdfs://master:9000";
	public static final Pattern DELIMITER = Pattern.compile("[\t,]");

	public static void main(String[] args) throws ClassNotFoundException,
			IOException, InterruptedException {
		sparseMartrixMultiply();
	}

	private static void sparseMartrixMultiply() throws ClassNotFoundException,
			IOException, InterruptedException {
		Map<String, String> path = new HashMap<String, String>();
		path.put("A", "/home/hadoop/logfile/matrix2/A.csv");// 本地的数据文件
		path.put("B", "/home/hadoop/logfile/matrix2/B.csv");
		path.put("input", HDFS + "/user/hdfs/matrix2");// HDFS的目录
		path.put("input1", HDFS + "/user/hdfs/matrix2/A");
		path.put("input2", HDFS + "/user/hdfs/matrix2/B");
		path.put("output", HDFS + "/user/hdfs/matrix2/output");

		MartrixMultiply.run(path);
	}

}


 
 
MartrixMultiply：
 

package com.oner.mr.sparsematrix;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
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 com.oner.mr.util.HdfsDAO;

public class MartrixMultiply {

	public static class SparseMatrixMapper extends
			Mapper<LongWritable, Text, Text, Text> {

		private String flag;// A or B;
		private int m = 4;// 矩阵A的行数
		private int p = 2;// 矩阵B的列数

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();
			flag = split.getPath().getName();// // 得到读取的矩阵名称
		}

		@Override
		protected void map(LongWritable k1, Text v1, Context context)
				throws IOException, InterruptedException {
			// 切分每行数据
			String[] fields = MainRun.DELIMITER.split(v1.toString());

			if ("A".equals(flag)) {
				for (int i = 1; i <= p; i++) {
					context.write(new Text(fields[0] + "," + i), new Text("A,"
							+ fields[1] + "," + fields[2]));
				}
			} else if ("B".equals(flag)) {
				for (int i = 1; i <= m; i++) {
					context.write(new Text(i + "," + fields[1]), new Text("B,"
							+ fields[0] + "," + fields[2]));
				}
			}

		}

	}

	public static class SparseMatrixReducer extends
			Reducer<Text, Text, Text, LongWritable> {
		@Override
		protected 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>();

			for (Text value : values) {
				String val = value.toString();

				if (val.startsWith("A")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到A,j,Aij中的j,Aij
					mapA.put(kv[0], kv[1]);// 将j作为key,Aij作为value存入mapA

				} else if (val.startsWith("B")) {
					String[] kv = MainRun.DELIMITER.split(val.substring(2));// 得到B,j,Bij中的i,Bij
					mapB.put(kv[0], kv[1]);// 将i作为key,Bij作为value存入mapB

				}
			}

			long result = 0;
			// 可能在mapA中存在在mapB中不存在的key，或相反情况
			// 因为，数据定义的时候使用的是稀疏矩阵的定义
			// 所以，这种只存在于一个map中的key，说明其对应元素为0，不影响结果
			Iterator<String> mkeys = mapA.keySet().iterator();// 得到mapA所有的键集合
			while (mkeys.hasNext()) {
				String mkey = mkeys.next();
				if (mapB.get(mkey) == null) {// 因为mkey取的是mapA的key集合，所以只需要判断mapB是否存在即可。
					continue;
				}
				result += Long.parseLong(mapA.get(mkey))
						* Long.parseLong(mapB.get(mkey));
			}
			context.write(key, new LongWritable(result));

		}
	}

	public static void run(Map<String, String> path) throws IOException,
			ClassNotFoundException, InterruptedException {

		Configuration conf = new Configuration();

		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("A"), input1);
		hdfs.copyFile(path.get("B"), input2);

		Job job = Job.getInstance(conf);
		job.setJarByClass(MainRun.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);
	}
}


 
 
打成jar包运行：hadoop jar matrix
 
 
查看结果：
 

[hadoop@master matrix2]$ hadoop fs -cat /user/hdfs/matrix2/output/part-r-00000
1,1	43
1,2	46
2,1	40
2,2	70
3,1	169
3,2	202
4,1	232
4,2	280