随着数据应用类型的愈加丰富和数据规模的不断扩大，单机早已无法满足超大规模数据计算和分析的需求，分布式才是如今大数据领域的核心关键词。作为数据分析领域非常优秀的Python计算库，Numpy、Scipy能够非常高效地执行单机数值计算，但却无法支持分布式计算，难以满足对海量数据进行分析的需求。为了解决这一痛点，涌现了一批为分布式场景下大规模数据科学计算而生的新计算库/计算引擎，比如Numba、Dask、谷歌开源的JAX、阿里开源的Mars等。

阿里的Mars自开源之初，就经常被拿来与Dask做对比，或被看作Numpy的替代者，那么Mars与Dask到底有何不同？Mars的终极目标真的是要取代Numpy吗？同时，当前有向无环图和分布式已经有Spark作为后端了，为什么还需要Mars？近日，我们有幸采访了阿里巴巴技术专家、Mars主要开发者之一斯文骏老师，对上述问题一一作了解答。此外，他还将在QCon 上海 2019带来《Mars：大规模张量计算系统的构建实践》的演讲分享，对Mars感兴趣的同学敬请关注。

Mars从何而来？

InfoQ：你们一开始决定开发Mars的初衷是什么？Mars和您之前参与开发的PyODPS这个项目之间有什么渊源或关联吗？（据了解，这两个项目的主要负责人是同一个人）

斯文骏：我们最初开发PyODPS的目的是为了方便数据开发者使用MaxCompute，其出发点和Mars是一致的，都是能让数据开发者使用与Python通用Library（Numpy、Scipy、Pandas）类似的用法来使用MaxCompute。但在开发和推广PyODPS的过程中，我们发现PyODPS存在不少局限性。PyODPS仅仅在用户端执行，其工作原理是将用户表达式编译为SQL，因而对于矩阵运算和Pandas中的Index变换等操作存在实现困难。即便上述操作可以使用SQL编写，SQL Engine对其执行效率也难以保证。因此，我们决定从更底层出发，开发一套兼容Python生态要求的分布式Library，且支持在MaxCompute中运行，这便是Mars。

相比PyODPS只支持DataFrame，且DataFrame与Pandas DataFrame存在一定的差异。Mars 直接依照Numpy、Scipy 和Pandas 接口编写，因而更符合Python开发者的习惯。

InfoQ：为什么业界需要分布式的科学计算引擎？原来Python中非常有名的Numpy、Scipy这些科学计算库存在什么问题？对大规模数据做科学计算有哪些难点？

斯文骏：Numpy、Scipy等计算库能够非常高效地在单机上进行数值计算，这也是Python为什么能够在数据分析领域迅速增长的重要原因。但随着数据规模的不断扩大，现有单机库逐渐难以应对在如此规模的数据中进行数据分析的需求。而分布式计算库需要在传统科学计算库之上加上分布式计算引擎的能力，具体包括：

根据计算的特点对数据进行拆分。例如，矩阵乘法和SVD在分布式计算的场景中，对于数据切分的要求不同。进行数据拆分和物理执行图的建立都是分布式计算库的职责；
执行过程中，执行图常常深度较大，且Barrier较少，调度器需要利用这一特性以减少中途的IO和等待代价；
分布式计算引擎共有的要求，例如计算、IO的高效并行化，故障恢复，等等。

InfoQ：Mars架构设计的理念和原则是什么样的？它在性能、易用性、泛化性、可靠性等方面做了哪些权衡？

斯文骏：Mars需要实现的是细粒度图调度和执行。用户使用Mars类型编写代码后，将形成一张逻辑上的粗粒度图。该Graph在提交到Scheduler后，会形成一张细粒度图，此后细粒度图的节点将被分配到各个Worker中执行。

我们基于自己实现的Actor Model搭建分布式调度服务。Actor Model使用Gevent实现异步操作，并包装所有通信，包括进程间和机器间的调用。每个Mars服务都由多个进程组成，以减少Python GIL对执行效率的影响。为减少大块数据在Worker内各进程间的复制成本，Mars使用Arrow进行序列化，并使用Plasma Store在各进程间共享所需的数据。当Plasma Store充满时，会Spill一部分旧数据到磁盘以使计算能继续进行下去。

Mars目前支持Worker进程以及Worker两级Failover。当Worker进程Fail，Worker Daemon会重建进程，并重跑受影响的节点。当整个Worker Fail，Scheduler会收到通知，根据丢失的 Worker以及数据血缘关系确定需要重跑的节点，重新分配初始节点并将相关节点再次提交执行。

Mars的定位

InfoQ：Mars的很多接口格式跟Numpy是一致的，有人认为这是为了让开发者能够从Numpy无痛迁移到Mars，也有人把Mars看作并行化/分布式化的Numpy，Mars开源的官方文章中对二者的性能也做了直接比较，所以Mars的终极目标是替代Numpy吗？

斯文骏：Numpy是非常强大的Python数值计算库，也是Python数值计算的事实标准。Mars对Numpy的兼容并不意味着要替换Numpy，而是在大规模分布式计算的情景下为用户提供科学计算的能力，这一能力在单机上的执行仍然需要依赖Numpy、Scipy、Pandas或者API类似Numpy、Pandas的Cupy、Cudf等单机库。当然，Numpy并不是在所有情形下执行效率都是最优的，所以会有Numexpr/Numba/Jax等在Numpy基础上实现加速的Library，而使用Mars在某些场合下也能达到这样的效果。

InfoQ：Mars、Dask和Ray做的都是Python并行数据分析，能否跟我们详细地解读一下，在实现并行计算的原理或方法上三者有哪些不同？三者各有什么优势和劣势？

斯文骏：Mars和Dask都是使用Python编写的以离线数据分析为目标的分布式并行计算库，且都拥有和Numpy/Pandas相近的API。但Mars和Dask在设计思路上有明显的差异。

表达式和计算图方面，Mars构建了一整套Tensor/DataFrame表达方式，采用Protobuf/JSON记录计算图，这使得Mars可以使用非Python客户端提交作业，也可以方便地采用Numexpr/Cupy等库对不同运算符采用个性化的优化。而Dask则采用pickle序列化Python Dict及Python Function的形式，对客户端/服务端的Python及相关Library的版本一致性要求很高，同时优化难度也较高。

分布式框架方面，Mars实现了自己的轻量化Actor模型，并在此基础上搭建Scheduler和Worker。Mars支持多个Scheduler以降低单点负载，同时通过多进程减少Python GIL的影响。而Dask则使用线程模型，单机调度效率较低。目前Benchmark的结果，Mars单机执行效率全面高过Dask，分布式执行中的大规模矩阵乘法等作业执行效率可达到Dask的3倍以上。同时，Mars还支持进程和Worker级别的Failover，使作业执行更加可靠。

Ray主要是为增强学习开发，提供了一套非常灵活的高效率分布式执行框架，但其并未直接提供分布式DataFrame等支持，也没有提供图优化、调度策略以及磁盘Spill等支持，用户上手会比较困难。现有基于Ray的Modin等库虽然提供了DataFrame功能，但也存在计算规模较低等问题。

InfoQ：有人认为有向无环图和分布式已经有Spark作为后端了，为什么还需要Mars？Mars的设计有何优势？

斯文骏：Spark和Mars在调度方式上是有显著差异的。Spark的DAG是一种粗粒度图，两个节点内的各Partition根据Narrow或者Wide Dependency建立一对一的连接或者全连接。Spark根据Wide Dependency切分DAG为Stage，每个Stage需要依次执行。而Mars的DAG则是细粒度图，每个Chunk可以有自己的依赖。这就意味着Mars在执行过程中，可以更精细地控制每个Chunk的执行和数据释放，从而获得更大的并行度和更高的效率。

在很多场景中，全连接事实上是不必要的。以矩阵乘法为例，分布式矩阵乘法可分为Chunk相乘阶段和乘积累加阶段。在Chunk相乘阶段，Tensor a中的每个Chunk并不是与Tensor b中的所有Chunk相乘，其所需的连接个数仅与Tensor b的某个维度的大小相同。因而，在执行中，无需等待所有Chunk乘法执行完成，即可执行后续的加法，待加法完成后，加法所依赖的乘积即可被释放，用于其他计算。因而Mars所采用的细粒度调度可以有效减少IO和Stage等待的开销，从而拥有更高的执行效率。

InfoQ：能否帮我们理清Mars和TensorFlow、JAX、MXNet等机器学习框架之间的关系？未来Mars有没有可能应用到上述这些深度学习框架里？

斯文骏：Mars和TensorFlow/PyTorch/MXNet等机器学习框架的出发点不同。Mars的出发点在于解决数据分析的规模问题。TensorFlow等机器学习框架的关注点不在规模，而在如何方便地实现机器学习/深度学习算法。其提供的分布式功能需要用户自行指定各个节点的职能。为构建生态，Mars正在考虑引入上述框架，使用户可以使用Mars产生的数据进行后续的机器学习，并将结果回流到Mars。

JAX是一套Numpy加速库，Mars正在尝试引入以提高执行速度。

InfoQ：据了解，Mars是阿里MaxCompute平台的一部分，请问它在整个MaxCompute平台中扮演的是什么样的角色？和其他计算引擎组件之间是如何配合的？

斯文骏：Mars可以独立部署执行，也可以使用PyODPS通过MaxCompute调度执行，从MaxCompute表中读取数据，执行结果回流到表中。目前MaxCompute中运行Mars处于内测阶段。

开源以来Mars的新变化

InfoQ：自开源以来，Mars又做了哪些改进？能否具体介绍一下。

斯文骏：自从开源以来，Mars已经从0.1演进到0.2，目前主干代码为0.3。自开源以来的主要更改包括：

初步增加DataFrame支持，已支持一元、二元运算、merge/join、reduction、iloc等操作；
支持Worker级别Failover，可在个别Worker fail的情况下使作业正常完成执行；
支持Shuffle，并在此基础上增加对Tensor Fancy Indexing等复杂操作的支持；
支持Eager Mode，可避免用户手动调用Execute，方便调试；
Worker级别Storage重构，统一管理各级存储，未来将可扩展到GPU Memory以及其他存储介质；
支持使用Kubernetes部署。

InfoQ：在早前一篇文章中，你们曾说过Mars会完全以开源的方式运作，而不是简单把代码放出来。开源这大半年来，Mars项目的推广情况如何？公司内部和外部使用情况如何？现在有多少开发者参与到代码贡献中了？

斯文骏：Mars目前所有开发、CI和Code Review都在Github上进行，目前主要是5位阿里内部开发者参与Mars的开发。近一年来我们工作的重心主要在Mars本身的完善而非推广，阿里内部和外部的使用主要是试用性质。近期我们会加大推广的力度。

InfoQ：您认为目前Mars还存在哪些问题？未来改进的重点是什么？能否分享一下接下来Mars在技术和开源社区两个方面的规划和Roadmap。

斯文骏：技术上，Mars未来会继续增强对Numpy/Pandas API的支持，进一步提高执行效率，同时建立接口以方便与其他数据开发/机器学习工具交换数据。Mars也会加强社区建设，近期将提供一系列代码解析文章，并通过建立开发者讨论组等形式加强开发者之间的联络。

采访嘉宾：斯文骏，阿里巴巴计算平台事业部技术专家，硕士研究生毕业后加入阿里巴巴，先后参与机器学习平台和 PyODPS 开发，2017 年参与启动 Mars 开发，为 Mars 分布式引擎主要开发者之一。

在QCon上海2019的演讲中，斯文骏老师将基于 Mars 以往的实践介绍系统架构和提升执行效率方面的实践，点击了解详情。

创作场景

为什么已有 Spark 和 Dask，阿里还要开源自研分布式科学计算引擎 Mars？