Spark上矩阵运算库（三）代码重构与优化

迁往MLlib的API

Spark的MLlib库中提供了分布式矩阵的API，本着不重复造轮子的原则，目前项目将代码重构使用MLlib提供的矩阵API，这些API都在org.apache.spark.mllib.linalg.distributed 包内，Spark的官方文档有一个简单的相关介绍：http://spark.apache.org/docs/latest/mllib-basics.html

本项目主要使用的API是 IndexedRowMatrix 和
IndexedRow ，其中IndexedRow是 (Long, Vector)的封装，两个参数分别是索引index和矩阵一行向量Array[Double]，而IndexedRowMatrix则由RDD[IndexedRow]实例化得到。

More spark-way-progaming

众所周知，Spark推荐使用函数式编程语言scala进行开发，而且由于涉及内存方面的迭代操作，所以即使同样能得到相同的结果，程序写得好坏会对程序执行时间、效率有着相当大的影响，目前我重构了代码，修改了算法，完全抛弃了collect()的相关action，程序执行的效果在下一节阐明，这也是我将该部分称之为more spark-way-programing 的原因。当然，如何更加优雅地写出scala代码，是一门技术还需要我日后不断摸索。

程序测试结果对比

在3个节点的Spark1.0.0上，程序采用spark-submit提交，执行代码如下：

$ bin/spark-submit --class edu.nju.matrixlib.FileMatrixMultiply --master spark://slave021:60123 \ --executor-memory 10G ../SparkProjects/matrixLibV2/target/scala-2.10/matrixlibv2_2.10-1.0.jar hdfs://slave021:22000/a.10000.10000.matrix
hdfs://slave021:22000/b.10000.10000.matrix tenthousand 5

程序stage的截图如下：





可以看出，两个分别是10^4 * 10^4 维度的矩阵相乘（运算复杂度是 10^12，每个存储矩阵的文本文件大小为860MB），只需要23分钟的时间，我将同样的数据在同样配置的单节点上调用Jama包测试，需要4个小时之久。除了运算是并行的，我们在从hdfs上读取和写入文件时，都是并行的。

下一步进展展望

1. 大集群测试实验

在3个节点上的测试基本已经结束，我已经在18个节点上部署了Spark on Yarn ，如下图：





下面将在更大的集群上测试更大的数据量，以及同时对比HAMA在该集群上的进行矩阵运算的情况

2. 继承扩展MLlib的矩阵相关API

由于MLlib所提供的相关矩阵API很多还处于@Experimental状态，而且提供的API都比较少，为了能够日后的使用方便，需要对MLlib这些类进行扩展，一个现在非常明显的问题就是在最后一步调用savaAsTextFile时，对于RDD的每一个元素会调用 toString 方法存储到hdfs上，而目前存储的每行格式是IndexedRow[1,(…,…,…)]，而我们希望的格式是1:…,…,…

3. 调用breeze库进行本地矩阵乘法

在mllib中，矩阵本地相乘使用的是breeze（https://github.com/scalanlp/breeze），我们这里本地矩阵相乘目前还是直接相乘，此处可以作为一个优化点。

4. shuffle过程和中间数据的优化

为了最后的数据存储方便，我们会生成大量的中间数据，这无疑增加了磁盘压力和网络传播，这需要进一步优化代码和程序架构设计。

5. 其他基本线性代数的实现

密集矩阵的相乘是一个难点，相较之下，其他诸如两个矩阵想加、相减等代数方法要简单不少，这是下一步需要实现的。