内存计算网格解释

Dmitriy Setrakyan在最近为In-Memory数据网格 （IMDG）提供了一个很好的解释 - 现在我尝试为In-Memory Compute Grid（IMCG）提供一些类似的描述。

IMCG - 内存计算网格

Dmitriy提出的主要想法之一便是内存存储（IMDG）和内存处理（IMCG）两者相互集成的重要性，这能够方便构建真正可扩展的应用程序。然而 - IMCG及其执行似乎并不如IMDG来的频繁，主要是由于下面的一些历史上的原因。

目前大多数供应商都会首要关注存储技术方面（如IMDG，NoSQL或NewSQL之类）。一旦建立了存储产品 - 就会更难在其上面添加任何类型的非基本IMCG功能，即使这并不是不可能（我们会在后面做出解释）。因此总的来说，IMCG功能对于整个产品来说更为基础，因此必须将其第一个构建或一起进行构建，以便在存储侧的核心中使用。

另外，GridGain和Hadoop仍然是市场上唯一能够成功地将存储和处理结合在一个产品中的产品（尽管存在很大差异），而现在仅有几十个存储项目可用（如果您统计过GitHub上的NoSQL尝试数量的话，可能达数百个）。

核心概念

理解IMCG最简单的方法是将其与IMDG进行比较。IMDG关注于通过把数据分布在网格之间可用的计算机中从而解决数据分布式内存存储和大数据集管理，而IMCG更专注于在同一网格上的同一组计算机上有效地执行算法（即用户代码或指令）。这就是两者的区别所在：IMDG是存储和管理内存中的数据，而IMCG则是处理和计算相同数据的全部内容。

从这个有利的角度看 - 很明显为什么IMDG和IMCG之间的紧密整合会如此重要：它们就像是同一个硬币的两个面 - 存储和处理，这两个面都会与您的数据相结合。

IMCG的大多数功能都可以分成独立的四个组：

1. 分布式部署和配置
2. 分布式资源管理
3. 分布式执行模型（又名IMCG Breadth）
4. 分布式执行服务（又名IMCG Depth）

1.分布式部署和配置

从历史上看，网格计算被认为是最尴尬和麻烦的重要原因之一便是将用户代码部署和配置到网格上执行，而且它在最糟糕的情况下甚至无法使用。从Globus，Grid Engine，DataSynapse，Platform Computing等早期产品到今天的Hadoop和大部分NoSQL项目 - 您需要手动部署和重新部署您的更改，包括重建所有库，将它们复制到任何地方，并重新启动您的服务。有些系统会为你复制并重新启动（如Hadoop），当然，也有些系统会要求你通过一些基于UI的拐杖手动执行。

由于IMCG是一种分布式技术，并且常常用于由数十甚至数百台计算机组成的拓扑结构，所以这个问题会更加严重。由此，在开发过程中停止服务，重新部署库并重新启动服务，CI测试以及在这些拓扑中进行升级便成为了一个主要问题。

GridGain是第一个通过提供“零部署”功能来减轻这个问题的IMCG。通过“零部署”，所有必需的JVM类和资源都会按需加载。此外，GridGain还提供了三种不同的对等部署模式，并且支持最复杂的部署环境，如自定义类加载器，WAR / EAR文件等。

零部署技术能让用户简单地将默认GridGain节点与这些节点联机，然后无需任何显式部署用户类或资源，并立即成为数据和计算网格拓扑的一部分，存储任何用户对象或执行任何用户任务。

2.分布式资源管理

分布式系统中的资源管理通常指管理物理设备（如计算机，网络和存储）以及软件组件（如JVM，运行时和操作系统）的能力。无论IMCG是否部署在某种受管理的基础设施上（如AWS），亦或是它是如何管理DevOps等，在不同情况下会有明显差别。

在IMCG中，自动发现和维护一致拓扑（即计算节点集合）是最重要的资源管理功能之一。自动发现允许用户在运行时从IMCG拓扑中添加和删除计算节点，同时保持IMCG上运行的任务不停机。一致的拓扑可确保所有计算节点都能以相同的顺序和一致性下查看任何拓扑更改（节点失效和离开或是新节点加入）。

GridGain提供了IMCG中最复杂的发现系统。可插拔和用户自定义的Discovery SPI是GridGain为其节点提供全自动和一致的发现功能的核心。GridGain提供了几种开箱即用的实现，包括基于IP多播和TCP / IP的实现，并能够直接支持AWS S3和Zookeeper。

3.分布式执行模型（又名）IMCG Breadth

IMCG是一种计算框架，支持不同的分布式执行模型。为了清晰起见，在执行模型（如MapReduce）和可以使用该模型实现的特定算法（即分布式搜索）之间有着明确的区别：有一组有限的执行模型，但实际上是一组无限可能的算法。

通常，IMCG（以及任何一般的计算框架）的目标是尽可能多地支持不同的执行模型，并为最终用户提供最多的选项，以便实现特定的算法并最终在分布式环境中执行。这就是我们经常称之为IMCG Breadth的原因。

例如，GridGain的IMCG能为以下执行模型直接提供支持：

• MapReduce的处理

GridGain提供针对内存优化的通用分布式fork-join类型处理。更具体地说，MapReduce类型处理定义了一个将原始计算任务分解为多个子任务的方法，即在任何托管基础设施上并行执行子任务，并将结果聚合（也就是减少结果）然后返回到最终结果之中。

GridGain的MapReduce本质上是一个分布式计算范例，它允许您将任务映射到基于某个键的较小作业上，并在网格节点上执行这些作业，然后将多个作业结果聚合到一个任务结果之中。这实际上是GridGain的MapReduce所完成的。但是，GridGain MapReduce与其他MapReduce框架（如Hadoop）的不同之处在于GridGain MapReduce适用于流式低延迟内存处理而后者不能。

如果Hadoop MapReduce任务从磁盘获取输入数据（input），在磁盘上生成中间结果并将结果输出到磁盘，则GridGain会负责处理Hadoop在内存中执行的所有操作 - 它直接调用API从内存中获取输入，然后在内存中生成中间结果，最后创建结果内存。完整的内存处理允许GridGain在短时间内提供结果，反观其他MapReduce框架则需要几分钟的时间。

• 流处理&CEP

流处理和相应的复杂事件处理（CEP）是一种处理类型，其中输入数据不是静态的，而是不断地“流”到系统中。其他MapReduce框架会产生不同的外部可执行进程，这些进程使用磁盘文件中的数据并将输出数据（output）输出到磁盘文件（在流模式下工作时也是如此），而GridGain Streaming MapReduce则与之不同，它可以直接在内存中直接处理流数据。

在数据进入系统的时候，用户可以继续生成MapReduce任务，并将它们分发到并行处理数据的远程节点集合处，并将结果返回给调用者。它的主要优点是所有MapReduce任务都直接在内存中执行，并且可以使用GridGain内存中的缓存来输入和存储结果，因此也拥有很低的延迟。

• MPP / RPC处理

GridGain还为传统的MPP（大规模并行处理）和RPC（远程过程调用）类型的处理提供本地支持，包括直接远程关闭执行，单播/广播/减少执行语义，共享分配会话和其他许多的功能。

• MPI风格的处理

GridGain的高性能分布式消息传递提供了MPI风格（即基于消息传递的分发）的处理能力。基于专有异步IO和世界上最快的编组算法，GridGain在分布式环境中提供了同步和异步语义，分布式事件和pub-sub消息。

• AOP / OOP / FP / SQL集成处理

GridGain是将计算网格功能集成到现有编程范例（如AOP，OOP，FP和SQL）的唯一平台：

• 您可以使用AOP为网格上的自动MapReduce或MPP执行注释Java或Scala代码。
• 您可以使用OOP和纯FP API来进行代码的MapReduce / MPP / RPC执行。
• GridGain允许将可执行的闭包注入到SQL执行计划中，允许您将自己的过滤器，本地和远程缩减器直接注入到ANSI SQL中。

3.分布式执行服务（又名IMCG Depth）

在许多方面，分布式执行服务是众所周知的执行模型“骨骼”中的“肉”。执行服务涉及许多支持各种执行策略和模型的深IMCG功能，如分布式故障转移，负载平衡，冲突解决等服务模型 - 因此它们被叫做IMCG Depths。

许多此类的功能在不同的IMCG和通用计算框架之间共享 - 但一些功能只针对于特定产品。以下是由GridGain的IMCG提供的一些关键执行服务的简短列表：

• 可插入的故障转移

故障转移管理和结果容错是任何网格计算基础架构都具有的关键属性。GridGain基于其SPI-based的架构，提供了完全可插拔的故障转移逻辑和几种可用的现成可用的实现。与其他网格计算框架不同，GridGain能够对逻辑进行故障转移，而非仅仅是数据。

在网格任务成为网格执行的最小单元的情况下，完全可定制和可插入的故障转移逻辑能使开发人员像在RDBMS事务中选择并发策略一样选择特定的策略。

而且，GridGain允许针对所有任务，或者一组任务甚至单独任务来定制故障转移逻辑。开发人员也可以使用元编程技术为每个任务执行定制故障转移逻辑。

而这也允许微调网络任务对故障的反应，下面举一些例子：

- 在其任何的一个工作失败时立即关闭整个任务（fail-fast 方法）

- 在拓扑耗尽之前将失败的工作通过故障转移到其它的节点（fail-slow 方法）

• 可插拔拓扑解析

GridGain提供了特定功能去直接或自动选择那些将执行MapReduce任务的网格节点子集（即拓扑）。这种能力让开发人员能够更灵活地决定任务执行的位置。而这个决定可以基于任何用户或系统信息。例如，一天的某个时刻或一周中的某一天，任务的类型，网格上的可用资源，来自给定节点的当前统计信息或平均统计信息，还有来自节点子集的集合，网络延迟，预定义的SLA等。

• 可插入的资源匹配

在某些网格节点比其他网格节点功能更强大或拥有资源更多的时候，您会遇到节点未充分利用或过度利用的情况。对于网格来说，无论是利用率低还是过度利用都一样很糟糕 - 理想情况下，网格中的所有网格节点都应该平等利用。GridGain提供了多种方式来实现整个网格的平等利用，例如：

加权负载平衡

如果事先知道某些节点比其他节点的能力强大2倍，则可以将比例权重附加到这些节点上。例如，一部分网格节点的权重为1，另一部分的网络节点权重为2。在这种情况下，作业的分布将与不同节点的权重成正比，权重较大的节点将等比例地被分配更多的作业，而权重较少的节点则反之。因此，权重为2的节点将获得比权重为1的节点多2倍的作业。

自适应负载平衡

如果不同节点之间不相同而您又不知道它们之间的不同之处，GridGain会帮您自动适应负载和处理能力的差异，并将更多的工作发送给能力更强大的节点，将更少的工作发送给更弱的节点。并且GridGain会监听各种节点上的各种指标并不断调整其负载平衡策略以适应负载的差异，从而实现自适应负载平衡。

• 可插入的冲突解决方案

冲突解决方案允许调整网格作业到达目标节点执行时的方式。类似于Mac OS X上的可定制GCD（Great Central Dispatch）任务管理，它允许开发人员自定义在单个节点上的作业分派。通常情况下，将会有多个作业在网格节点上执行，同时也有可能是多个作业正在执行或等待执行。这时有多种可能的策略去处理这种情况，例如所有作业并行进行，或者作业被序列化，或者在任何给定的时间内只执行一个作业，或者只有特定数量或类型的网格作业可以并行执行等...

• 可插拔的早期和晚期负载平衡

对于通过负载平衡和冲突解决SPI定义的Compute Grid，GridGain能为其提供早期和晚期负载均衡 - 它能有效实现整个负载均衡过程的完全自定义。早期和晚期负载平衡允许将网格任务执行调整为在网格上执行的非确定性性质。

早期负载平衡由MapReduce进程的映射操作提供支持。映射 - 将作业映射到解析拓扑中的节点的过程 - 在任务执行开始时发生，因此它被认为是早期的负载平衡。

一旦作业被调度到远程节点执行，它们就会排队在远程节点上启动。这项工作在队列中的等待时长以及何时执行都由冲突SPI来进行控制，这有效地定义了后期负载平衡阶段。

作业窃取算法是负载均衡业务流程的一种开箱即用的实现。它将在后期检测不平衡状况，并在作业在实际执行之前从繁忙节点发送到空闲节点。

网格和云环境通常是异构和非静态的，任务可以在运行时动态地改变其复杂性配置文件，并且外部资源可以在任何时候影响到任务的执行。所有这些因素都强调了在初始映射操作期间以及作业可能处于等待队列的目标节点时，主动负载平衡需求。

• 分布式任务会话

分布式任务会话是为了每个任务的执行而产生，而且它允许在任务内的不同作业之间共享状态。作业可以添加，获取和等待各种属性设置，并且网格作业和任务可以保持连接状态，以便彼此同步执行，并为解决全新的问题提供解决方案。

想象一下，如果你需要压缩一个非常大的文件（比方说大小为TB）。如果要在网格环境中执行此操作，您可以将此文件分成多个部分，并将每个部分分配给远程作业执行。每项作业都需要扫描其所分配的部分以寻找是否有内容重复。一旦这项扫描由所有作业并行完成，作业需要将结果与其同胞（siblings）同步，以便在整个文件中一致地进行压缩。而这可以通过设置每一个作业发现的重复内容来实现。

• 冗余映射支持

在某些情况下，保证及时的成功结果比执行冗余作业重要得多。这时，GridGain允许您在MapReduce任务中产生同一作业的多个副本，以在远程节点上并行执行。每当第一个作业成功完成时，其他相同的作业将被取消和忽略。这种方法可以在牺牲冗余执行的条件下，更好地保证成功及时地完成工作。只要您的电网不会过载并且实现冗余所消耗的CPU成本并不昂贵，请使用冗余映射支持。

• 节点本地缓存

当在分布式环境中工作时，通常需要在每个网格节点上具有一致的本地状态，并在各种作业执行之间重复使用。例如，如果多个作业需要数据库连接池执行 - 如何让这个连接池初始化一次，然后在同一个网格节点上运行的所有作业重新使用？从本质上讲，你可以把它看作是一个单独的网格节点单例服务，但是这个想法并不仅限于服务，它可以只是一个普通的Java bean，它拥有一些状态以便在同一个网格节点上运行的所有作业共享。

• 基于Cron的调度

除了在整个网格或网格（虚拟子网格）的任何用户定义部分上直接运行MapReduce任务之外，还可以根据需要安排任务重复运行。GridGain支持基于Cron的任务调度语法，因此您可以使用我们熟悉的标准Cron语法来安排和运行您的任务。

• 部分异步减少

有时，执行MapReduce任务时，您无需等待所有远程作业全部完成，即可完成任务。一个很好的例子就是简单搜索。例如，假设您正在从多个远程节点上的GridGain数据网格中缓存的数据中搜索某种模式。一旦第一份工作返回时发现模式，你就不需要等待其他工作完成，因为你已经找到了你正在寻找的东西。对于像这样的情况，GridGain允许您在收到来自远程作业的所有结果之前减少（或完成）您的任务 - 因此名称为“部分异步减少”。在这种情况下，您网格中任务的剩余工作将被取消。

• 可插入任务检查点

检查任务提供了定期保存状态的功能。而这与故障转移功能结合起来尤其有用。设想一个可能需要执行5分钟的作业，但4分钟后，它所在的节点就会崩溃。作业将故障转移到另一个节点，但必须从头开始重新启动，并且依旧需要5分钟。但是，如果作业每分钟都有检查点，那么可能丢失的工作大部分是执行时的最后一分钟，并且在故障转移后，作业将从最后保存的检查点重新开始。GridGain允许您轻松地进行检查点作业，以更好地控制作业和任务的总体执行时间。

• 分布式连续性

对于需要暂停工作并需要释放资源的情况，延续很有用。例如，如果从作业中产生新任务，则等待该任务同步完成是错误的，因为作业线程在等待期间将保持占用状态，因此网格中的线程可能会被用尽。正确的做法是暂停作业，以便稍后（例如在每当新生成的作业完成之时）继续作业。这是GridGain延续真正有用​​的地方。GridGain允许用户在任何时候暂停和重新开始工作。因此，在我们的示例中，远程作业需要产生另一个任务并等待结果，我们的作业会产生任务执行，然后暂停自己本身。紧接着，每当新任务完成时，我们的工作就会醒来并恢复执行。这种方法允许简单的任务嵌套和递归的任务执行。它还允许您在系统中拥有比可用线程多得多的交叉相关作业和任务。

• 与IMDG整合

与基于亲和性路由的IMDG集成是计算和数据网格技术背后的关键概念之一（无论是内存还是基于磁盘）。通常，关联路由允许共同定位作业和该作业需要处理的数据集。

这个想法非常简单：如果作业和数据不在同一地点，则作业将到达某个远程节点，并且必须从存储数据的另一个节点获取必要的数据。一旦处理完毕，这些数据很可能会被丢弃（因为它已经存储并在其他地方备份）。这个过程导致了昂贵的网络旅程加上所有相关的编组和解组。在规模上 - 这种行为几乎可以使任何系统停下来。相似性协同定位通过将作业与其必要的数据集共同定位解决了这个问题。我们说处理（即作业）和处理需要的数据之间存在相似性 - 因此，我们可以根据此关联性将作业发送到存储数据的节点，以避免不必要的网络出行和额外的编组以及解组处理。GridGain提供了高级功能以实现亲和力共位：从简单的单一方法调用到支持复杂亲和键和非平凡拓扑的复杂API。

例

下面的例子演示了一个典型的无状态计算任务，即在网格上用Pi编号计算（用Scala编写 - 不过也可以用Java或Groovy或Clojure轻松完成）。这个例子展示了GridGain的实现有多么简单 - 而且只有十几行代码。

请注意，这是一个完整的源代码 - copy'n'paste，编译并运行它。同时还要注意的是，它可以在一个节点上工作 - 并且也可以在网格或云中的一千个节点上工作，而且无需更改代码 - 只是呈线性地更快。更有趣的是，这个应用程序自动包含所有这些执行服务：

• 自动拓扑发现
• 自动负载平衡
• 分布式故障转移
• 碰撞解决
• 零代码部署和供应
• 可插拔编组和通信

Scala代码：

import org.gridgain.scalar._
import scalar._
import scala.math._

object ScalarPiCalculationExample {
    private val N = 10000

    def main(args: Array[String]) {
        scalar {
            println("Pi estimate: " +
                grid$.spreadReduce(for (i <- 0 until grid$.size()) yield () => calcPi(i * N))(_.sum))
        }
    }

    def calcPi(start: Int): Double =
        // Nilakantha algorithm.
        ((max(start, 1) until (start + N)) map 
            (i => 4.0 * (2 * (i % 2) - 1) / (2 * i) / (2 * i + 1) / (2 * i + 2)))
            .sum + (if (start == 0) 3 else 0)
}