漫谈分布式计算框架

如果问 mapreduce 和 spark 什么关系，或者说有什么共同属性，你可能会回答他们都是大数据处理引擎。如果问 spark 与 tensorflow 呢，就可能有点迷糊，这俩关注的领域不太一样啊。但是再问 spark 与 MPI 呢？这个就更远了。虽然这样问多少有些不严谨，但是它们都有共同的一部分，这就是我们今天谈论的一个话题，一个比较大的话题：分布式计算框架。

不管是 mapreduce，还是 spark 亦或 tensorflow，它们都是利用分布式的能力，运行某些计算，解决一些特定的问题。从这个 level 讲，它们都定义了一种“分布式计算模型”，即提出了一种计算的方法，通过这种计算方法，就能够解决大量数据的分布式计算问题。它们的区别在于提出的分布式计算模型不同。Mapreduce 正如其名，是一个很基本的 map-reduce 式的计算模型（好像没说一样）。Spark 定义了一套 RDD 模型，本质上是一系列的 map/reduce 组成的一个 DAG 图。Tensorflow 的计算模型也是一张图，但是 tensorflow 的图比起 spark 来，显得更“复杂”一点。你需要为图中的每个节点和边作出定义。根据这些定义，可以指导 tensorflow 如何计算这张图。Tensorflow 的这种具体化的定义使它比较适合处理特定类型的的计算，对 tensorflow 来讲就是神经网络。而 spark 的 RDD 模型使它比较适合那种没有相互关联的的数据并行任务。那么有没有一种通用的、简单的、性能还高的分布式计算模型？我觉着挺难。通用往往意味着性能不能针对具体情形作出优化。而为专门任务写的分布式任务又做不到通用，当然也做不到简单。

插一句题外话，分布式计算模型有一块伴随的内容，就是调度。虽然不怎么受关注，但这是分布式计算引擎必备的东西。mapreduce 的调度是 yarn，spark 的调度有自己内嵌的调度器，tensorflow 也一样。MPI 呢？它的调度就是几乎没有调度，一切假设集群有资源，靠 ssh 把所有任务拉起来。调度实际上应当分为资源调度器和任务调度器。前者用于向一些资源管理者申请一些硬件资源，后者用于将计算图中的任务下发到这些远程资源进行计算，其实也就是所谓的两阶段调度。近年来有一些 TensorflowOnSpark 之类的项目。这类项目的本质实际上是用 spark 的资源调度，加上 tensorflow 的计算模型。

当我们写完一个单机程序，而面临数据量上的问题的时候，一个自然的想法就是，我能不能让它运行在分布式的环境中？如果能够不加改动或稍加改动就能让它分布式化，那就太好了。当然现实是比较残酷的。通常情况下，对于一个一般性的程序，用户需要自己手动编写它的分布式版本，利用比如 MPI 之类的框架，自己控制数据的分发、汇总，自己对任务的失败做容灾（通常没有容灾）。如果要处理的目标是恰好是对一批数据进行批量化处理，那么 可以用 mapreduce 或者 spark 预定义的 api。对于这一类任务，计算框架已经帮我们把业务之外的部分（脚手架代码）做好了。同样的，如果我们的任务是训练一个神经网络，那么用 tensorflow pytorch 之类的框架就好了。这段话的意思是，如果你要处理的问题已经有了对应框架，那么拿来用就好了。但是如果没有呢？除了自己实现之外有没有什么别的办法呢？

今天注意到一个项目，Ray，声称你只需要稍微修改一下你的代码，就能让它变为分布式的（实际上这个项目早就发布了，只是一直没有刻意关注它）。当然这个代码仅局限于 python，比如下面这个例子，

| **Basic Python**                               | **Distributed with Ray**                           |
+------------------------------------------------+----------------------------------------------------+
|                                                |                                                    |
|  # Execute f serially.                         |  # Execute f in parallel.                          |
|                                                |                                                    |
|                                                |  @ray.remote                                       |
|  def f():                                      |  def f():                                          |
|      time.sleep(1)                             |      time.sleep(1)                                 |
|      return 1                                  |      return 1                                      |
|                                                |                                                    |
|                                                |                                                    |
|                                                |  ray.init()                                        |
|  results = [f() for i in range(4)]             |  results = ray.get([f.remote() for i in range(4)]) |
+------------------------------------------------+----------------------------------------------------+

这么简单？这样笔者想到了 openmp（注意不是 openmpi）。来看看，

#include
#include"omp.h"

using namespace std;

void main() {
#pragma omp parallel for
    for(int i = 0; i < 10; ++i) {
        cout