基于Spark实现的超大矩阵运算

2024-05-12 23:48

本文主要是介绍基于Spark实现的超大矩阵运算,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

由于标题强调了是在Spark平台实现的矩阵运算,所以本文会非常有针对性的介绍,甚至细节到Spark RDD的算子。

1.算法描述

思想其实很简单,就是矩阵分块计算,而分块矩阵就成了小矩阵,然后就借助于Breeze实现。而对于Spark平台而言,其处理流程如下图:


2.矩阵分块依据

这里仅仅提供一种思路,所以仅供参考。假设有两个矩阵A和B,其中A是m*k的矩阵,B是k*n的矩阵,CPU的总核数是cores,则分块方法:

  • m > k && m > n --> m/2 && cores/2
  • k > m && k > n --> k/2 && cores/2
  • n > k && n > m --> n/2 && cores/2

3.分块矩阵ID标识:BlockID

由于BlockID最后要依靠RDD在集群中通信传输,所以BlockID必须是可序列化的。另外,BlockID要作为分块矩阵的唯一标识,所以BlockID必须具有唯一性,而BlockID的唯一由一下3个属性确定:

  • blockRow:表示该子/分块矩阵在原矩阵中的行号;
  • blockCol:表示该子/分块矩阵在原矩阵中的列号;
  • blockSeq:表示该子/分块矩阵的序列号,默认为0。

4.矩阵分块原理

由于Spark处理文件时,是一行一行的处理的,所以一开始读文件,构成的RDD的类型是:RDD[(seqnum, DenseVector)] (seqnum:输入的行号,DenseVector:对应seqnum的矩阵行)。同时,我们还需要知道2个数据:

  • allrow:矩阵的总行数
  • allcol:矩阵的总列数
另外,由于矩阵运算中,矩阵形状的不同,所以分块的方式也随之而异。如下图,左上图就需要按列分块,右上图就需要按行分块,左下图就需要行列都分块,右下图就需要分别按列分块和按行分块。


4.1按行分块,列不分块

这时需要知道以下2个数据:

  • rowblocknum:按行分块的数量
  • subrow:每块矩阵的行数
然后,分三步处理:

①mapPartitions{map}将RDD[(seqnum, DenseVector)]组成新的数据结构:RDD[(seqnum/subrow, (seqnum, DenseVector))]
②groupByKey作用RDD[(seqnum/subrow, (seqnum, DenseVector))]得到新的数据结构RDD[(seqnum/subrow, Iterable[(seqnum, DenseVector)])]

e.g.
allrow = 1000, rowblocknum = 5, subrow = allrow/rowblocknum = 200

③mapPartitions{map}把Iterable[(seqnum, DenseVector)]的数据填装到子/分块矩阵submatrix中
④构建新的数据结构:RDD[(BlockID, submatrix)]

4.2按行按列分块,和按列分块行不分

这时,我们需要知道3个数据,和准备一个存储行向量的数组:
  • element: Array 读入的每行数据
  • subcol: 每块矩阵的列数
  • colblocknum:按列分块的数量
  • arrayBuff: ArrayBuffer[(BlockID, (Long, Vector))] 存储按列切分的行向量
①mapPartitions{flatMap}将输入的每行数据按列切分,存储到arrayBuff: ArrayBuffer[(BlockID, (Long, Vector))]
②groupByKey作用RDD[(BlockID, (Long, Vector))]得到新的数据结构RDD[(BlockID, Iterable[(seqnum, DenseVector)])]
e.g.
allrow = 1000, rowblocknum = 5, subrow = allrow/rowblocknum = 200
allcol = 1000, colblocknum = 5, subcol = allcol/colblocknum = 200

③mapPartitions{map}把Iterable[(seqnum, DenseVector)]的数据填装到子/分块矩阵submatrix中
④构建新的数据结构:RDD[(BlockID, submatrix)]

5.矩阵乘法的例子

例如:有两个矩阵A和B,其中A是6m*4k的矩阵,被分为3*2块个子矩阵;B是4k*4n的矩阵,被分为2*2块的子矩阵。如图:


下标(x,y,z)是每个子/分块矩阵的唯一标识BlockID(row: Int, col: Int, seq: Int = 0)的参数,即:

  • x:表示该子/分块矩阵在原矩阵中的行号,即blockRow;
  • y:表示该子/分块矩阵在原矩阵中的列号,即blockCol;
  • z:表示该子/分块矩阵的序列号,默认为0,即blockSeq。
和分块块数:
  • mSplitNum:表示矩阵A按行切分的块数;
  • kSplitNum:表示矩阵A按列切分的块数,也是矩阵B按行切分的块数;
  • nSplitNum:表示矩阵B按列切分的块数。
对于该例子,mSplitNum=3、kSplitNum=2、nSplitNum=2。
①mapPartitions{flatMap}把RDD[(BlockID, submatrix)],即矩阵A的每个子/分块矩阵按序列号生成nSplitNum个RDD[(BlockID, submatrix)],矩阵B的每个子/分块矩阵按序列号生成mSplitNum个RDD[(BlockID, subMatrix)],使之一一对应。
对于矩阵A
val array = Array.ofDim[(BlockID, DenseMatrix[Double])](nSplitNum)for (i <- 0 until nSplitNum) {val blockSeq = blockRow * nSplitNum * kSplitNum + i * kSplitNum + blockColarray(i) = (new BlockID(blockRow, i, blockSeq), DenseMatrix)
}

对于矩阵B

val array = Array.ofDim[(BlockID, DenseMatrix [Double])](mSplitNum)for (i <- 0 until mSplitNum) {val blockSeq = i * nSplitNum * kSplitNum + blockCol * kSplitNum + blockRowarray(i) = (new BlockID(i, blockCol, blockSeq), DenseMatrix)
}

e.g. mSplitNum=3,kSplitNum=2,nSplitNum=2
MatrixA

MatrixB


即:MatrixA每个子/分块矩阵复制nSplitNum份,MatrixB每个子/分块矩阵复制mSplitNum份,然后把Key值BlockID相同的子/分块矩阵相乘。
②join两矩阵A和B,使每一对subMatrix相乘,同时更新BlockID(blockRow, blockCol)使blockSeq默认为0。
③reduceByKey按BlockID把子/分块矩阵的乘积相加,得到最终的矩阵。


声明:这只是个人思想,仅做参考。按照这个想法,如果不做任何优化(比如,相乘的小矩阵不分块,而是采用广播的方式等等),在我的实验集群中好像最多能处理10000*10000*10000规模的数据集。


参考文献:

http://www.open-open.com/doc/view/dc6d0ce0233d4db397fd677a2d0a55dc

这篇关于基于Spark实现的超大矩阵运算的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/984081

相关文章

python生成随机唯一id的几种实现方法

《python生成随机唯一id的几种实现方法》在Python中生成随机唯一ID有多种方法,根据不同的需求场景可以选择最适合的方案,文中通过示例代码介绍的非常详细,需要的朋友们下面随着小编来一起学习学习... 目录方法 1:使用 UUID 模块(推荐)方法 2:使用 Secrets 模块(安全敏感场景)方法

Spring StateMachine实现状态机使用示例详解

《SpringStateMachine实现状态机使用示例详解》本文介绍SpringStateMachine实现状态机的步骤,包括依赖导入、枚举定义、状态转移规则配置、上下文管理及服务调用示例,重点解... 目录什么是状态机使用示例什么是状态机状态机是计算机科学中的​​核心建模工具​​,用于描述对象在其生命

Spring Boot 结合 WxJava 实现文章上传微信公众号草稿箱与群发

《SpringBoot结合WxJava实现文章上传微信公众号草稿箱与群发》本文将详细介绍如何使用SpringBoot框架结合WxJava开发工具包,实现文章上传到微信公众号草稿箱以及群发功能,... 目录一、项目环境准备1.1 开发环境1.2 微信公众号准备二、Spring Boot 项目搭建2.1 创建

IntelliJ IDEA2025创建SpringBoot项目的实现步骤

《IntelliJIDEA2025创建SpringBoot项目的实现步骤》本文主要介绍了IntelliJIDEA2025创建SpringBoot项目的实现步骤,文中通过示例代码介绍的非常详细,对大家... 目录一、创建 Spring Boot 项目1. 新建项目2. 基础配置3. 选择依赖4. 生成项目5.

Linux下删除乱码文件和目录的实现方式

《Linux下删除乱码文件和目录的实现方式》:本文主要介绍Linux下删除乱码文件和目录的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux下删除乱码文件和目录方法1方法2总结Linux下删除乱码文件和目录方法1使用ls -i命令找到文件或目录

SpringBoot+EasyExcel实现自定义复杂样式导入导出

《SpringBoot+EasyExcel实现自定义复杂样式导入导出》这篇文章主要为大家详细介绍了SpringBoot如何结果EasyExcel实现自定义复杂样式导入导出功能,文中的示例代码讲解详细,... 目录安装处理自定义导出复杂场景1、列不固定,动态列2、动态下拉3、自定义锁定行/列,添加密码4、合并

mybatis执行insert返回id实现详解

《mybatis执行insert返回id实现详解》MyBatis插入操作默认返回受影响行数,需通过useGeneratedKeys+keyProperty或selectKey获取主键ID,确保主键为自... 目录 两种方式获取自增 ID:1. ​​useGeneratedKeys+keyProperty(推

Spring Boot集成Druid实现数据源管理与监控的详细步骤

《SpringBoot集成Druid实现数据源管理与监控的详细步骤》本文介绍如何在SpringBoot项目中集成Druid数据库连接池,包括环境搭建、Maven依赖配置、SpringBoot配置文件... 目录1. 引言1.1 环境准备1.2 Druid介绍2. 配置Druid连接池3. 查看Druid监控

Linux在线解压jar包的实现方式

《Linux在线解压jar包的实现方式》:本文主要介绍Linux在线解压jar包的实现方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录linux在线解压jar包解压 jar包的步骤总结Linux在线解压jar包在 Centos 中解压 jar 包可以使用 u

c++ 类成员变量默认初始值的实现

《c++类成员变量默认初始值的实现》本文主要介绍了c++类成员变量默认初始值,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧... 目录C++类成员变量初始化c++类的变量的初始化在C++中,如果使用类成员变量时未给定其初始值,那么它将被