MPP架构和批处理

目录

数据库构架

MPP和批处理

MPP概念

MPP的设计缺陷

将MPP和Batch进行结合

MPP例子

 Hadoop解决的问题

MPP和Hadoop的区别

小结

数据库构架

数据库构架设计中主要有Shared Everthting、Shared Nothing、和Shared Disk：

Shared Everthting:一般是针对单个主机，完全透明共享CPU/MEMORY/IO，并行处理能力是最差的，典型的代表SQLServer

Shared Disk：各个处理单元使用自己的私有 CPU和Memory，共享磁盘系统。典型的代表Oracle Rac， 它是数据共享，可通过增加节点来提高并行处理的能力，扩展能力较好。其类似于SMP（对称多处理）模式，但是当存储器接口达到饱和的时候，增加节点并不能获得更高的性能 。

Shared Nothing：各个处理单元都有自己私有的CPU/内存/硬盘等，不存在共享资源，类似于MPP（大规模并行处理）模式，各处理单元之间通过协议通信，并行处理和扩展能力更好。典型代表DB2 DPF和hadoop ，各节点相互独立，各自处理自己的数据，处理后的结果可能向上层汇总或在节点间流转。

我们常说的 Sharding 其实就是Share Nothing架构，它是把某个表从物理存储上被水平分割，并分配给多台服务器（或多个实例），每台服务器可以独立工作，具备共同的schema，比如MySQL Proxy和Google的各种架构，只需增加服务器数就可以增加处理能力和容量。

MPP和批处理

批处理系统 – 使用场景分钟级、小时级以上的任务，目前很多大型互联网公司都大规模运行这样的系统，稳定可靠，低成本。

MPP系统 – 使用场景秒级、毫秒级以下的任务，主要服务于即席查询场景，对外提供各种数据查询和可视化服务。

MPP概念

MPP即大规模并行处理（Massively Parallel Processor ）。 在数据库非共享集群中，每个节点都有独立的磁盘存储系统和内存系统，业务数据根据数据库模型和应用特点划分到各个节点上，每台数据节点通过专用网络或者商业通用网络互相连接，彼此协同计算，作为整体提供数据 库服务。非共享数据库集群有完全的可伸缩性、高可用、高性能、优秀的性价比、资源共享等优势。

MPP DBMS是基于此方法构建的数据库管理系统，在这些系统中，您所注视的每个查询都被拆分为由MPP网格节点并行执行的一组协调进程，将计算拆分为比传统SMP RDBMS系统运行速度更快的方式。

MPP解决方案的最原始想法就是消除共享资源。每个执行器有单独的CPU，内存和硬盘资源。一个执行器无法直接访问另一个执行器上的资源，除非通过网络上的受控的数据交换。这种资源独立的概念，对于MPP架构来说很完美的解决了可扩展性的问题。

为了能够处理大量数据，这些解决方案中的数据通常按每个节点仅处理其本地数据的方式在节点之间拆分（分片）。这进一步加快了数据的处理速度，因为将共享存储用于这种设计将是一个巨大的过大杀伤力–更复杂，更昂贵，扩展性更差，网络利用率更高，并行性更低。这就是为什么大多数MPP DBMS解决方案都是不共享的，并且不能在DAS存储或在小型服务器组之间共享的一组存储架上工作的原因

MPP的第二个主要概念就是并行。每个执行器运行着完全一致的数据处理逻辑，使用着本地存储上的私有数据块。在不同的执行阶段中间有一些同步点(我的理解：了解Java Gc机制的，可以对比GC中stop-the-world,在这个同步点，所有执行器处于等待状态)，这些同步点通常被用于进行数据交换(像Spark和MapReduce中的shuffle阶段)。这里有一个经典的MPP查询时间线的例子： 每个垂直的虚线是一个同步点。例如：同步阶段要求在集群中”shuffle”数据以用于join和聚合(aggregations)操作,因此同步阶段可能执行一些数据聚合，表join，数据排序的操作，而每个执行器执行剩下的计算任务。

MPP的设计缺陷

但是，这样的设计对于所有的MPP解决方案来说都有一个主要的问题——短板效应。如果一个节点总是执行的慢于集群中其他的节点，整个集群的性能就会受限于这个故障节点的执行速度(所谓木桶的短板效应)，无论集群有多少节点，都不会有所提高。这里有一个例子展示了故障节点(下图中的Executor 7)是如何降低集群的执行速度的。

大多数情况下，除了Executor 7 其他的所有执行器都是空闲状态。这是因为他们都在等待Executor 7执行完成后才能执行同步过程，这也是我们的问题的根本。比如，当MPP系统中某个节点的RAID由于磁盘问题导致的性能很慢，或者硬件或者系统问题带来的CPU性能问题等等，都会产生这样的问题。所有的MPP系统都面临这样的问题。

如果你看一下Google的磁盘错误率统计报告，你就能发现观察到的AFR(annualized failure rate,年度故障率)在最好情况下，磁盘在刚开始使用的3个月内有百分之二十会发生故障。

如果一个集群有1000个磁盘，一年中将会有20个出现故障或者说每两周会有一个故障发生。如果有2000个磁盘，你将每周都会有故障发生，如果有4000个，将每周会有两次错误发生。两年的使用之后，你将把这个数字乘以4，也就是说，一个1000个磁盘的集群每周会有两次故障发生。

事实上，在一个确定的量级，你的MPP系统将总会有一个节点的磁盘队列出现问题，这将导致该节点的性能降低，从而像上面所说的那样限制整个集群的性能。这也是为什么在这个世界上没有一个MPP集群是超过50个节点服务器的。

MPP和批处理方案如MapReduce之间有一个更重要的不同就是并发度。并发度就是同一时刻可以高效运行的查询数。MPP是完美对称的，当查询运行的时候，集群中每个节点并发的执行同一个任务。这也就意味着MPP集群的并发度和集群中节点的数量是完全没有关系的。比如说，4个节点的集群和400个节点的集群将支持同一级别的并发度，而且他们性能下降的点基本上是同样。下面是我所说情况的一个例子。

正如你所见，10-18个并行查询会话产生了整个集群最大的吞吐量。如果你将会话数提高到20个以上的时候，吞吐量将慢慢下降到70%甚至更低。在此声明，吞吐量是在一个固定的时间区间内(时间足够长以产生一个代表性的结果)，执行的相同种类的查询任务的数量。Yahoo团队调查Impala并发度限制时产生了一个相似的测试结果。Impala是一个基于Hadoop的MPP引擎。因此从根本上来说，较低的并发度是MPP方案必须承担的以提供它的低查询延迟和高数据处理速度。

将MPP和Batch进行结合

我们现在可以看到两个架构的优点和短板。MPP是更快的，但是有两个关键痛点——短板效应和并发限制。而对于像MapReduce这样的批处理系统，我们需要花费时间来存储中间数据到磁盘上，但与此同时，我们获得了更高的扩展度而因此可以获得远远大于MPP的集群规模。我们如何才能将两者结合来获得MPP低延迟和高速处理，使用batch-like的设计来降低短板效应和并发度低的问题？我想如果我告诉你答案是新的Apache HAWQ的架构你是不会惊讶的。

再一次提出问题，MPP的查询是如何执行的？通过一定数量的平行执行的进程运行完全相同的代码，进程数目和集群的节点数量是完全一致的，在每个节点上处理本地数据。但是，当我们介绍HDFS的时候，你不会把数据和本地Executor绑定在一起，这也就意味着你可以摆脱Executor数目的限制，你也就不需要在固定的节点上处理本地存在的数据(在传统MPP上，你不能处理远程节点的数据).为什么？因为HDFS默认对同样的block存储3个备份，也就意味着集群中至少有3个节点上，你可以选择创建一个Executor并处理本地的数据。并且，HDFS支持远程读取数据，也就意味着至少有两个机架上可以处理本地机架上的数据，这样就可以通过最少的拓扑数来远程获取数据。

这也就是为什么Apache HAWQ提出了”virtual segments”的概念——GreenPlum中的”segment” 是改进版的PostgreSQL数据库中的一个单一实例，它在每个节点上存在一个，并且在每次查询中产生”executor”进程。如果你有一个小的查询，它可以被4个executors执行甚至是一个。如果你有一个大的查询，你可以用100个甚至1000个executor执行。每个查询仍然是以MPP风格对本地数据进行处理，而且不需要将中间数据写入到HDD上，但是”virtual segments”允许executor运行在任何地方。下面是它的一个工作示例图（不同颜色代表了不同的查询，虚线代表了查询中的shuffle操作）

这赋予了你以下的特性：

[1] 减轻MPP系统的短板问题：因为我们可以动态的添加节点和删除节点。因此，严重的磁盘故障将不会影响整个集群的性能，系统可以拥有比传统MPP更大量级的集群。现在，我们可以暂时的将一个故障节点从集群中移除，那么就不会有更多的executor在上面开始运行。并且，移除节点时不会有停机时间。

[2] 一次查询现在被一个动态数量的executors进行执行，这也就带来了更高的并发度，缓和了MPP系统的限制并加入了batch系统的灵活性。想象一下拥有50个节点的集群，每个节点最多可以运行200个并行的进程。这就意味着你一共拥有了”50*200=10,000”个”execution slot”。你可以对20个查询每个执行500个executor，也可以对200个查询每个执行50个executor，甚至于你可以对1个查询运行10000个executor。在这里是完全灵活可控的。你也可能有一个大的查询要使用4000个segments和600个小的查询每个需要10个executors，这也是可以的。

[3] 数据管道的完美应用：实时的从一个executor中将数据转移到另一个executor中。在执行阶段，单独的查询仍然是MPP的，而不是batch。因此，不需要将中间数据存储到本地磁盘中(无论何时，操作都允许数据管道)。这意味着我们离MPP的速度更近一步了。

[4] 像MPP那样，我们仍然尽可能的使用本地数据来执行查询，这一点可以通过HDFS的short-circuit read(当client和数据在同一节点上，可以通过直接读取本地文件来绕过DataNode，参考HDFS Short-Circuit Local Reads)来实现。每个executor会在拥有该文件最多块数的节点上创建并执行，这也最大化了执行效率。

MPP例子

Greenplum是一种基于PostgreSQL的分布式数据库。其采用shared nothing架构（MPP），主机，操作系统，内存，存储都是自我控制的，不存在共享。也就是每个节点都是一个单独的数据库。节点之间的信息交互是通过节点互联网络实现。通过将数据分布到多个节点上来实现规模数据的存储，通过并行查询处理来提高查询性能。

这个就像是把小数据库组织起来，联合成一个大型数据库。将数据分片，存储在每个节点上。每个节点仅查询自己的数据。所得到的结果再经过主节点处理得到最终结果。通过增加节点数目达到系统线性扩展。

elasticsearch也是一种MPP架构的数据库，Presto、Impala等都是MPP engine，各节点不共享资源，每个executor可以独自完成数据的读取和计算，缺点在于怕stragglers，遇到后整个engine的性能下降到该straggler的能力，所谓木桶的短板，这也是为什么MPP架构不适合异构的机器，要求各节点配置一样。

Spark SQL应该还是算做Batching Processing, 中间计算结果需要落地到磁盘，所以查询效率没有MPP架构的引擎（如Impala）高。

 Hadoop解决的问题

Hadoop存储技术是基于完全不同的方法构建的。它不是基于某种密钥对数据进行分片，而是将数据分块成固定（可配置）大小的块，并在节点之间进行拆分。块很大，而且它们是只读的，也是整个文件系统（HDFS）的。简单地说，将100行的小表加载到MPP中会导致引擎根据表的键对数据进行切分，这样在一个足够大的集群中，每个节点只存储一行的可能性很大。相反，在HDFS中，整个小表将被写入一个单独的块中，在datanode的文件系统中它将被表示为一个文件。

与MPP设计相比，Hadoop resource manager（YARN）为您提供了更细粒度的资源管理—与MPP相比，MapReduce作业不需要所有计算任务并行运行，因此，如果集群的另一部分被完全利用，您甚至可以在单个节点上运行的一组任务中处理大量数据。它还具有一系列很好的特性，如可扩展性、支持长寿命容器等。但事实上，它比MPP资源管理器慢，有时在管理并发性方面不太好。

像Impala和HAWQ这样的解决方案就在这个边缘的另一边，它们是Hadoop之上的MPP执行引擎，处理HDFS中存储的数据。与其他MPP引擎一样，它们可以为查询提供更低的延迟和更短的处理时间，同时降低可伸缩性和稳定性：

SparkSQL是介于MapReduce和MPP与Hadoop方法之间的另一头猛兽，它试图从两个世界中获得最好的结果，但也有自己的缺点。与MR类似，它将工作分成一组单独安排的任务，从而提供更好的稳定性。与MPP一样，它尝试在执行阶段之间传输数据以加快处理速度。此外，它还使用MPP（及其impalad和HAWQ段）所熟悉的固定执行器概念来减少查询的延迟。但是它也结合了这些解决方案的缺点——没有MPP那么快，没有MapReduce那么稳定和可伸缩。

MPP和Hadoop的区别

小结

MPP和Batch架构，正在逐渐走向融合，Batch架构确实在实时响应性方面是缺点，同样的MPP架构由于扩展性和木桶效应导致集群规模不能很大。

如果对扩展性有着更高要求，可以选择Batch架构，如果你希望提供更快的查询选择MPP架构。我不想过多讨论更多HAWQ(历史包袱)，因为SQL on Hadoop已经出现太多失败的项目，而HAWQ的出现给我们提供了新思路和方向，Batch+MPP融合架构。目前基于Hadoop发展起来的各种Batch/Stream/MPP系统都在努力解决问题，为未来新型数据分析方法提供更多样化的解决方案。

包括一个很敏感的话题，Hadoop上云，我们慢慢讨论。

就因为忍受不了Batch系统，我才面带领团队研究DataFlow产品，解决Batch面对今天日益严峻的实时性数据分析响应问题，DataFlow产品会完全取代ETL，让数据一直在流，流动中被处理，并且被高效处理、实时反馈。

我在Clickhouse似乎看到的新型OLAP数据仓库的未来，虽然还有很多缺陷和问题需要解决完善.

目前OLAP分布式计算引擎系统发展的方向：

Batch    MapReduce -> Spark/Tez

Stream   SparkStreaming -> Flink Streams -> Kafka KSQL

Batch + MPP  Greenplum/HP Vertica -> Dremel -> Impala/Drill/PrestoDB/HAWQ

Real-time Storage(Search/KV)  Druid.io/ElesticSearch/CrateDB/Hbase

目前一分钟上百万数据实时入库，实时查询系统属于Real-time Storage(Search/KV)场景。