如何进行Mars的应用与实践

59次阅读

共计 3583 个字符，预计需要花费 9 分钟才能阅读完成。

这篇文章给大家介绍如何进行 Mars 的应用与实践，内容非常详细，感兴趣的小伙伴们可以参考借鉴，希望对大家能有所帮助。

一、Mars 简介

Mars 是统一的数据科学平台，它用来加速传统的 Python 数据科学技术栈，在单机中也可以用多核加速，或用分布式来加速。Mars 可以部署在单机的分布式集群，或者 Kubernetes 和 Hadoop Yarn 上。

Mars 整个框架构建在单机的并行和分布式的这两个调度的基础之上，它的数据科学基础包括三个核心部分，Tensor、DataFrame 和 Remote。而构建在这个基础之上的，是 Mars Learn 模块，它可以兼容 Scikit-learn API，能简单地进行更大数据规模的分布式处理。此外，Mars 还支持深度学习和机器学习的框架，比如能轻松运行 TensorFlow、PyTorch 等，而且可视化也可以在 Mars 上完成。除此之外，Mars 还支持了丰富的数据源。

从传统 Python 技术栈到 Mars 也非常简单，比如在 NumPy 和 Pandas 里要变成 Mars，只需要替换 import，然后后面变为延迟执行即可。

普通的 Python 函数，在调用的时候变成 mr.spawn 来延迟这个过程，最后通过 execute 来并发执行，不用担心 Mars 是运行在单机上还是分布式运行。

而 Mars 上的 TensorFlow 大部分也一样，区别在于 main 函数部分的变化。最后，我们需要通过 run_tensorflow_script 的方式把脚本运行到 Mars 中。

二、典型场景

场景 1. CPU 和 GPU 混合计算

在安全和金融领域可以用 Mars 做 CPU 和 GPU 的混合计算，加速现有的工作流。

在这个领域，因为传统大数据平台挖掘周期长，资源紧张，需要很久来执行任务，不能达到客户需求。所以能用 Mars DataFrame 来加速数据处理，它可以做大规模数据排序，并帮助用户做高阶统计和聚合分析。

另外在安全领域有很多无监督学习的算法，Mars learn 能加速无监督学习，同时拉起分布式深度学习计算加速现有的深度学习训练。之后，对于某些计算任务也可以利用 GPU 来加速。

场景 2. 可解释性计算

在广告领域，在广告归因和洞察特征的解释算法中，因为本身计算量大，所以耗时很长。这种情况下，单机加速是比较困难的，基于传统大数据平台的分布式也不太灵活，但是通过 Mars remote，可以很轻松地把计算分布到几十台机器上来加速，达到百倍的性能提升。

场景 3. 大规模 K - 最邻近算法

Mars 非常广泛地应用在 K - 最邻近算法中，因为 Embedding 越来越流行，它让向量表述实体非常常见。另外，Mars 的 NearestNeighbors 算法兼容 scikit-learn，它里面有暴力算法，而用户也需要暴力算法来进行大规模计算，可以通过多个 worker 来完成，从而让性能提升百倍。最后，Mars 支持分布式的方式加速 Faiss 和 Proxima，达到千万级别和上亿级别的规模。

三、Demo

Demo1. 分析豆瓣电影数据

我们从这个 Demo 看一下 Mars 如何加速 pandas 数据处理及其可视化。

开始演示之前我们需要安装 Mars。这里已经创建了 Jupyter，然后 pip install pymars。

安装之后，我们可以到 IPython 进行验证，可以看到下面的结果没有问题，接下来我们就可以进入到 Jupyter notebook 里。

我们开始 demo。这个数据可以在 GitHub 地址下载，然后我们用 pandas 来分析电影的数据，使用 ipython memory usage 来查看内存使用。

我们的数据主要用到 4 个 CSV 文件，分别是 movies、ratings、users 和 comments。

接下来根据上映日期统计有多少电影发布。这里先处理一下数据，让发行日期只取到年份，去掉日期，并对年份做聚合。

数据出来后，可以用 pandas bokeh 把图绘制出来，并通过交互式的方式查看。

接下来看电影评分的统计。首先把有评分的电影筛选出来，然后把豆瓣评分的数值数量从大到小进行排序。可以看到，最多的评分是 6.8 分。

同样，通过 pandas bokeh 把它画成柱状图，评分差不多呈现正态分布。

接下来做一个标签词云，看电影哪个标签词最多，这里从 movies 取出 tags，用斜杠分割，然后 max words 是 50。

接下来我们再对电影的 Top K 进行分析。首先按电影 ID 进行聚合，求出评价的平均值和个数。然后我们对评价个数进行过滤，从高到低，算出 top20 的电影。

然后做评论数据分析。因为评论是中文的，所以需要做一个分词，然后对每一句话做切分，在统计的时候进行排序。这里可以加一个进度条，在处理数据的时候方便看到进程。这个过程大概花了 20 分钟，所以在单机上跑大任务的时候对机器的压力还是比较大。

这是最终的词云图。

接下来我们用 Mars 做同样的分析任务。首先是对 Mars 环境进行部署，然后这里有 5 个 worker，每个 worker 是 8 个 CPU 和 32G 内存。还是一样，我们打开内存的监控，做一些 import，这里把 import Pandas 替换成 import mars.dataframe，然后 Numpy 是 import mars.tensor。

随后我们在 SDK 里来创建 to mars dataframe，这一步几乎没有用到内存，最终得到的结果也和之前一样。

我们用同样的方式来分析上映日期的电影个数和电影评分。得益于 Mars 跟 Pandas 的高度兼容，我们也能用 Pandas bokeh 来呈现结果。

电影评论的分析也一样，但是在显示的时候，Mars 只会拉取头几条和最后几条，所以客户端几乎没有内存使用。而且整个 running 过程只用了 45 秒，与之前的 20 分钟相比提升了几十倍性能。

接下来我们用 Mars 做一个地区的统计，让它有一个动态的效果。首先我们看一下刚刚计算过的已经 released 的电影 dataframe，然后取 1980-2019 这几年的电影，而 regions 部分可能有多个，所以用斜杠分割开，最后执行排出 top10 地域电影。

然后我们通过 bar chart race 来生成动态效果。

Demo2. 豆瓣电影推荐

第二个 demo 我们会基于刚才豆瓣电影的数据来做一个推荐。我们首先会用 TensorFlow Mars 来进行训练，接着用 Mars 分布式 KNN 算法来加速召回计算。

我们先使用单机的技术栈，这个数据已经分成了训练和测试集，所以我们先 to pandas 把它下载到本地，接着来对用户和电影做一个 label encode，把它变成一个数字，而不是字符串的值。随后我们对数据进行处理，先按照时间排序，然后按照用户进行分组，生成分组聚合的结果。

接下来开始训练，我们需要用 TensorFlow 训练出代表 user 的 embedding。之前说过 embedding，可以对任一实体用向量描述，所以得到 embedding 之后，我们在给用户推荐电影时就可以查找在这个向量空间里面跟这个用户比较接近的电影 embedding。

训练后我们可以保存向量，这里的搜索规模是 60 万乘 7 万，单机花费了 22 分钟，但如果达到千万乘千万级别，搜索耗时要超过 800 小时，这是不可接受的。

接下来我们看如何用 Mars 来实现这一过程。首先创建一个 Mars 集群，这里有 8 个 worker。然后和上面一样，对数据进行预处理，做 label encode，按时间排序，按 user 分组生成分组聚合。

这里唯一的区别是 Mars 会自动推断 DataFrame 的结果，如果推断失败就需要用户自己提供 dtypes 和 output type。

然后是执行和训练。这里 TensorFlow 可以写 Python 文件，不用写到 notebook 里。

接着我们用 Mars 的 run tensorflow script 来跑这个脚本，然后指定 worker 是 8。可以看到，执行的时间缩小到了 23 分钟。同时，我们也拿到了最终的 embedding，用 Mars 做 embedding 只需 1 分 25 秒，比刚刚的时间提升个十倍左右。1400 万乘 1400 万也可以稳定在 1 小时左右，与单机 800 个小时相比提升是非常巨大的。

四、最佳实践

首先尽量不要使用 to pandas 和 to numpy，因为这会把 Mars 的分布式数据变成单机的数据，失去了 Mars 本身的优势，除非这个操作不能用 Mars 实现；其次，Mars tensor、DataFrame 和 learn 由于本身受限于 API 的原因需要自己写一些函数，所以可以考虑用 Mars remote 来加速，把操作抽象成函数；第三，Pandas 的加速技巧在 Mars DataFrame 依然适用，比如可以使用更高效的数据类型，可以优先使用内建操作，使用 apply 取代循环。