# 网络大数据 - spark分析

对于矩阵乘法$A\times B$，我们在Spark实现中是采用笛卡尔积的方式完成的（即将两矩阵$A$、$B$分别视作行的集合和列的集合，每一行和每一列对应到结果中的一个元素）：

因为`sim`和`items`都是`Dataset[_]`类型，其数据是可以有具体类型的（类似于结构体），Spark在实现笛卡尔积时会先将它们序列化为一个统一的内部结构`Row`（准确说是`InternalRow`，类比数据库中的“行”，或者“元组”），在笛卡尔积处理结束后再将其反序列化为具体的结构体（原因大概是笛卡尔积是用无条件JOIN操作实现的，而JOIN操作一般来说是对元组进行的操作）。

通过`queryExecution`可以看到原始的logical plan如下图所示：

而经过优化后最终生成的physical plan仍然保留了反序列化这一操作（在我们的例子中反序列化和序列化都是冗余操作，但是logical plan的优化策略只识别到了序列化的冗余性，并将序列化操作删除，而没有处理反序列化操作）：

这里`DeserializeToObject`实际执行时运行[代码](https://github.com/apache/spark/blob/d8920ae13346670bac3f1a7e49dfc2ba85c93b03/sql/core/src/main/scala/org/apache/spark/sql/execution/objects.scala#L95-L101)如下：

其中`GenerateSafeProjection`会[生成](https://github.com/apache/spark/blob/0494dc90af48ce7da0625485a4dc6917a244d580/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateSafeProjection.scala#L175-L198)`SpecificSafeProjection`类：

并嵌入如下[代码](https://github.com/apache/spark/blob/0494dc90af48ce7da0625485a4dc6917a244d580/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateSafeProjection.scala#L106-L110)，其行为是复制整个数组：

上述完整制造数据拷贝的代码段，在我们的实验场景下会带来严重的性能开销。说明如下：
- 矩阵$A$、$B$的规模分别为$n\times m$、$m\times m$，其中$n$和$m$在同一数量级，矩阵$B$可以视作稠密矩阵，矩阵$A$则在每一行中只有$k$个元素（$k \ll m$）
- 不考虑数据复制的代价，矩阵乘法时间开销只有$O\left(k \times m \times m\right) \approx O\left(m^2\right)$
- 而考虑数据复制的代价，笛卡尔积中每个结果项都需要复制大小为$O\left(m+k\right)$的数据，由于结果项的数量是$O\left(m^2\right)$，故额外引入的时间开销可达$O\left(m^3\right)$

在程序实际运行时通过`perf`工具，对函数热点进行分析，上述代码片段（`SpecificSafeProjection.apply`方法）占用总时间开销的接近乃至超过一半，验证了上述分析是成立的：