用于实时分析的开源数据存储

摘要

Druid 是一个开源数据存储，专为基于大数据集的实时探索和分析而设计。该系统包括列式存储、分布式无共享架构和复杂的索引结构，可用于以亚秒级延迟任意探索10 亿行数据表。本文主要介绍Druid的架构，并详细介绍其如何支持快速聚合、灵活过滤、低延迟数据加载。

1. 引言

近年来，随着互联网技术的快速发展，机器生成的事件数量迅速增加。单独来看，这些事件并没有包含太多有用的信息，价值不大。考虑到从大量事件中提取意义所需的时间和资源，许多公司宁愿放弃这些数据。正在构建基础设施来处理基于事件的数据，但其中大部分售价很高，并且仅适用于有能力负担得起的企业。

几年前，Google 引入了MapReduce 作为一种使用商用硬件进行Web 索引和日志分析的机制。不久之后，Hadoop 项目就继承了原始MapReduce 文档的见解，并在很大程度上对其进行了模仿。 Hadoop 目前被许多组织用来存储和分析大量日志数据。 Handoop 帮助许多公司将低价值事件流转变为高价值应用程序，例如商业智能和A-B 测试。

与许多出色的系统一样，Hadoop 为我们打开了一个新的问题空间。具体来说，Hadoop在存储和提供对大量数据的访问方面表现出色，但它并不能保证数据访问速度方面的性能。此外，尽管Hadoop 是一个高可用系统，但高并发负载会降低性能。最后，虽然Hadoop 非常擅长存储数据，但它并未针对提取数据和使其易于读取进行优化。

在Metamarket产品开发的早期，我们就遇到了这些问题，并意识到Hadoop是一个优秀的后端处理、批处理和数据仓库系统。然而，作为一家需要在高并发环境（1000+用户）下对查询性能和数据可用性进行操作级保证的公司，Hadoop并不能满足我们的需求。在考虑了各种解决方案并尝试了关系数据库管理系统和NoSQL 架构之后，我们得出的结论是，开源世界中没有任何东西可以充分满足我们的需求。我们最终创建了Druid，一个开源、分布式、面向列的实时分析数据存储。在很多方面，Druid 与其他OLAP 系统、交互式查询系统、内存数据库和众所周知的分布式数据存储类似。分布式和查询模型还从现代搜索基础设施中获得见解。

本文介绍了Druid 的架构，讨论了创建支持托管服务的始终在线生产系统所需的各种设计决策，并可能为面临类似问题的人提供指导，我们将尽力帮助您找到解决方法。 Druid 已被多家科技公司用于生产。本文的结构如下。我们首先在第2 节中描述问题。接下来，第3 节从数据如何流经系统的角度详细介绍了系统架构。接下来，第4 节解释了如何以及为何将数据转换为二进制格式。第5 节简要描述了查询API，第6 节介绍了性能结果。最后，我们将在第7 节介绍运行德鲁伊的课程，并在第8 节介绍相关任务。

2. 问题定义

Druid 最初设计是为了解决摄取和探索大量事务事件（日志数据）的问题。这种形式的时间序列数据常见于OLAP 工作流程中，并且数据量通常非常大。

例如表1中的数据。表1 列出了有关维基百科上所做编辑的数据。每次用户编辑维基百科页面时，都会生成一个事件，其中包含有关编辑的元数据。该元数据由三个不同的组件组成。首先，有一个时间戳列告诉您编辑的时间。接下来，有配置维度列，显示有关编辑的各种属性，例如编辑的页面、进行编辑的用户以及用户的位置。最后，还有一组度量列，其中包含可以聚合的值（通常是数字），例如在编辑中添加或删除的字符数。

我们的目标是快速计算这些数据的下钻和聚合。我们还回答了诸如“旧金山人贾斯汀·比伯对他的页面进行了多少更改？”和“卡尔加里人每月添加的平均字符数是多少？”之类的问题。对于任意维度组合的查询返回延迟小于1 秒。

Druid 的出现是因为当前的开源关系数据库RDBMS 和NoSql 键/值存储无法为交互式应用程序提供低延迟的数据摄取和查询平台。在Metamarkets 的早期，我们专注于构建托管仪表板，使用户能够自由探索和可视化事件流。为仪表板提供支持的数据存储必须足够快地返回查询，以便基于其构建的数据可视化可以为用户提供交互式体验。

除了查询延迟要求之外，系统还必须是多租户且高度可用的。 Metamarkets 产品旨在用于高并发环境。停机成本高昂，许多企业无法等待软件升级或面临导致系统不可用的网络故障。对于往往缺乏适当内部运营控制的初创企业来说，停机时间可以直接决定业务的成败。

最后，Metamarket 的另一个早期挑战是使用户和警报系统能够“实时”做出业务决策。事件创建到可查询之间的时间决定了利益相关者对系统中潜在灾难性情况做出响应的速度。流行的开源数据仓库系统（例如Hadoop）无法提供所需的亚秒级数据摄取延迟。

数据探索、摄取和可用性问题涉及多个行业。 Druid 于2012 年10 月开源，此后已被多家公司部署为视频、网络监控、运营监控和在线广告分析平台。

3. 架构

Druid 集群由不同类型的节点组成，每种节点类型都旨在执行一组特定的操作。我们相信这种设计功能可以解耦并简化整个系统的复杂性。不同类型的节点彼此独立运行，节点之间的交互被最小化。因此，集群内通信故障对数据可用性的影响也被最小化。

为了解决复杂的数据分析问题，不同类型的节点协同工作，形成一个功能齐全的系统。德鲁伊这个名字来源于很多角色扮演游戏中的角色德鲁伊。德鲁伊是变形者，能够呈现不同的形态来履行团体中的不同角色。图1 显示了Druid 集群内的数据组织和流程。

3.1 Real-time Node

实时节点封装事件流摄取和查询功能。通过这些节点索引的事件可立即用于查询。节点仅考虑短时间范围内的事件，并定期将在这段短时间范围内收集的不可变批次事件传递到Druid 集群中专门处理不可变事件批次的其他节点。活动节点利用Zookeeper 与Druid 集群的其余部分进行协调。节点将其在线状态和数据传递给Zookeeper 服务。

实时节点在内存中为所有传入事件维护一个索引缓冲区。这些索引是随着事件的摄取而逐渐建立的，并且索引是可直接查询的。当查询驻留在基于JVM 堆的缓冲区中的事件时，Druid 的行为类似于行存储。为了避免堆溢出问题，活动节点会定期或在达到最大行限制后将内存中的索引持久保存到磁盘。这个持久过程将存储在内存缓冲区中的数据转换为第4 节中描述的列式存储格式。每个持久索引都是不可变的，活动节点将持久索引加载到堆外内存中，以便可以查询。该过程在[33]中有详细描述，并如图2 所示。

每个活动节点都会定期调度一个后台任务来搜索所有本地持久索引。该任务组合这些索引并构建一个不可变的数据块，其中包含实时节点在一段时间内摄取的所有事件。该数据块称为“段”。在切换阶段，活动节点将段上传到持久备份存储，通常是分布式文件系统，例如S3 [12] 或HDFS [36]（Druid 称之为“深度存储”）。摄取、持久化、合并和切换步骤是流畅的，并且在任何过程中都不会丢失数据。

图3显示了实时节点的运行情况。该节点从13:37开始，只接受当前时间或下一次的事件。当事件被摄取时，节点宣布它正在提供13:00 到14:00 区间内的数据段。每10 分钟（周期可配置），节点刷新其内存缓冲区并将其保留在磁盘上。接近时间结束时，节点上可能会发生从14:00 到15:00 的事件。发生这种情况时，节点准备在下一小时提供数据并创建新的内存索引。然后，该节点宣布它还为从14:00 到15:00 的段提供服务。该节点不会立即合并持久索引13:00 到14:00，而是等待直到从13:00 到14:00 的可配置事件窗口到达。此窗口期可最大限度地降低因事件传递延迟而导致数据丢失的风险。在窗口结束时，节点将从13:00 到14:00 的所有持久索引合并到单个不可变段中，并移交该段。当该段在Druid 集群中的其他位置被加载和查询时，活动节点会清除其收集的有关13:00 到14:00 数据的所有信息，并停止对该数据的广告服务。

3.1.1 可用性和可扩展性

实时节点是数据的消费者，需要相应的生产者提供数据流。通常，出于数据持久化的目的，生产者和实时节点之间使用诸如kafka [21] 之类的消息总线，如图4 所示。实时节点通过从消息总线读取事件来摄取数据。从事件创建到事件消费的时间通常约为几百毫秒。

3.2 Historical Nodes

历史节点依赖于Zookeeper 的段加载和卸载指令。如果Zookeeper 不可用，历史节点将无法提供新数据或丢弃旧数据。但是，由于查询是通过HTTP 提供的，因此历史节点仍然可以响应对其当前提供的数据的查询请求。这意味着Zookeeper 中断不会影响历史节点上当前数据的可用性。

3.2.1 层

代理节点充当历史节点和实时节点的查询路由器。 Broker 节点了解Zookeeper 中公开的元数据，并且知道它们可以查询哪些段以及这些段所在的位置。代理节点路由传入的查询，以便它们到达正确的历史或实时节点。代理节点还将历史节点和实时节点的部分结果进行合并，并将最终合并结果返回给调用者。

3.2.2 可用性

代理节点包含具有无效LRU（最近最少使用）策略的缓存。缓存可以使用本地堆内存或外部分布式键/值存储（例如Memcached）。每当代理节点收到查询时，它首先将查询映射到一组段。有些段结果可能已经存在于缓存中，因此不需要重新计算。对于缓存中没有的结果，代理节点将查询转发到正确的历史节点和实时节点。当历史节点返回结果时，代理节点将这些结果逐段缓存以供将来使用。此过程如图6 所示。实时数据不会被缓存，因此对实时数据的请求始终转发到实时节点。实时数据会永远变化，缓存的结果不可靠。

缓存还可以作为数据持久性的附加级别。如果所有历史节点都失败，如果结果已经在缓存中，您仍然可以查询结果。

3.3.2 可用性

即使与Zookeeper的通信中断，数据仍然可以查询。如果代理节点无法与Zookeeper 通信，代理节点将继续使用集群的最新已知视图将查询转发到实时节点和历史节点。代理节点假设集群的结构与中断之前相同。事实上，这种可用性模型允许您的Druid 集群在很长一段时间内继续处理查询，即使诊断出Zookeeper 中断也是如此。

3.3 Broker Nodes

Druid Coordinator节点主要负责History节点中的数据管理和分发。协调节点指示历史节点加载新数据、删除过期数据、复制数据以及将数据移至负载均衡器。 Druid 使用多版本并发控制交换协议来管理不可变段并维护稳定的视图。如果不可变段包含被新段完全覆盖的数据，则旧段将从集群中删除。协调器节点必须经过领导者选举过程来确定单个节点作为主节点来执行协调器功能，其余协调器节点作为冗余备份。

协调器节点定期运行以确定集群的当前状态。通过将运行时集群的预期状态与集群的实际状态进行比较来做出决策。与所有Druid 节点一样，协调器节点通过Zookeeper 连接维护当前集群信息。协调器节点还维护与包含其他操作参数和配置的MySQL 数据库的连接。 MySQL 中的重要信息之一是一个表，其中包含历史节点提供的所有段的列表。该表可以由创建该段的任何服务（例如活动节点）更新。 MySQL 数据库还包含规则表，用于管理如何在集群内创建、销毁和复制段。

3.3.1 缓存

规则确定历史段如何加载到集群中以及如何从集群中删除。规则指示如何将段分配给不同的历史节点层以及每个层中应存在段的多少个副本。规则还可以确定何时从集群中永久删除段。规则通常是在一段时间内制定的。例如，用户可以使用规则将最近一个月的段加载到“热”集群中，将最近一年的段加载到“冷”集群中，并删除旧的段。

协调器节点从MySQL 规则表加载一组规则。您可以专门为您的数据源配置规则或配置默认规则集。协调器节点循环遍历所有可用段，并将每个段与应用于其的第一个规则进行匹配。

3.4 Coordinator Nodes

在典型的生产环境中，查询通常会产生数十个甚至数百个段。由于每个历史节点的资源有限，因此协调器必须分布在集群中的节点之间，以防止集群变得过于不平衡。确定最佳负载分布需要对查询模式和速度有一定的了解。查询通常涵盖来自单个数据源的连续时间间隔的最新片段。平均而言，访问较小段的查询速度更快。

这些查询模式建议快速复制最近的历史片段，扩展不同历史节点上时间接近的大型片段，以及将来自不同数据源的片段放在一起。为了在整个集群中优化分配和平衡段，我们开发了一个基于成本的优化流程，该流程考虑了数据源、新近度和段的大小。该算法的具体细节超出了本文的范围，可能会在未来的文献中介绍。

3.4.1 规则

协调器节点可以指示不同历史节点加载同一段的副本。历史节点集群每层的副本数量是完全可配置的。需要高级别容错能力的设置可以配置大量副本。段副本的处理方式与原始段相同，并遵循相同的负载平衡算法。通过复制段，过去的单节点故障在Druid 集群中透明地发生。该属性用于软件升级。您可以无缝地使历史节点脱机、更新它们、备份它们，并对集群中的所有历史节点重复该过程。在过去的两年里，我们的Druid 集群从未因软件升级而出现任何停机情况。

3.4.2 负载均衡

Druid 协调器节点具有Zookeeper 和MySQL 作为外部依赖项。协调器节点依赖于ZooKee

per来确定集群中已经存在哪些历史节点。如果Zookeeper不可用，coordinator将不再能够发送指令来分配，平衡和丢弃segments。但是，这些操作不会影响数据可用性。
集群中存在哪些段的segment元数据信息，还是哪些段应该存在集群中的segment元数据信息
MySQL和Zookeeper故障响应的设计原则是相同的：如果负责协调的外部依赖失败，集群维持现状。Druid使用MySQL存储操作管理信息，关于集群中应存在哪些段的元数据信息。如果MySQL停止，此信息对协调器节点不可用。但是，这并不意味着数据本身不可用。如果协调器节点不能与MySQL通信，它们将停止分配新的段并丢弃过时的节点。在MySQL中断期间，Broker、历史和实时节点仍然可以查询。

4. 存储结构

Druid中的数据表（称为数据源）是时间戳事件的集合，并分割为一组segments，其中每个段通常为5-10万行。正式地，我们将段定义为跨越某个时间段的数据行的集合。段表示Druid中的基本存储单元，复制和分发都是在段级别完成的。
Druid总是需要一个时间戳列，用来简化数据分发策略，数据保留策略和第一级查询修剪。Druid数据源划分成定义良好的时间间隔（通常为一小时或一天），并且可以进一步对来自其他列的值进行分区，以实现所需的段大小。分割段的时间粒度是数据量和时间范围的函数。如果数据集中的时间戳遍布在一年里，则按天进行分区。如果数据集中的时间戳遍布在一天里，则按小时进行分区。
段由数据源标识符进行唯一标识，标识符包括数据的时间间隔以及新段被创建时增加的版本字符串。版本字符串可以识别出段数据的新鲜度；新版本的段具有较新的数据视图（在一些时间范围内）。该段元数据由系统用于并发控制; 读操作总是从具有该时间范围的最新版本标识符的段中访问特定时间范围内的数据。
Druid段用列式存储。鉴于Druid最适合用于事件流的聚合计算（所有进入Druid的数据必须有一个时间戳），所以将聚合信息存储为列而不是行的优势已有详细记录[1]。列存储允许更高效的利用CPU，因为只有实际需要的才会被加载和扫描。在面向行的数据存储器系统中，与行相关联的所有列必须作为聚合的一部分进行扫描。额外的扫描时间可以引入明显的性能退化[1]。
Druid有多种列类型来表示各种数据格式。根据列类型不同，使用不同的压缩方法来降低在内存和磁盘上存储列的成本。在表1中给出的示例中，page、user、gender和city列仅包含字符串。直接存储字符串需要不必要的代价，可以用字典编码来代替。字典编码是压缩数据的常用方法，并已用于其他数据存储系统，如PowerDrill [17]。在表1的示例中，我们可以将每个page映射到唯一的整数标识符。

Justin Bieber -> 0
Ke$ha -> 1
此映射允许我们将page列表示为整数数组，其中数组索引对应于原始数据集的行。对于page列，我们可以表示为：[0, 0, 1, 1]
生成的整数数组本身非常适合进行压缩。在编码之上的通用压缩算法在列存储中非常常见。Druid使用LZF [24]压缩算法。
类似的压缩方法可以应用于数字列。例如，表1中characters added和characters removed列也可以表示为单个数组：
Characters Added -> [1800, 2912, 1953, 3194]
Characters Removed -> [25, 42, 17, 170]
在这种情况下，我们压缩原始值，而不是它们的字典表示。

4.1 数据过滤索引

在许多现实世界的OLAP工作流中，针对满足某些维度条件的某些度量集合的聚合结果发出查询。比如一个查询示例是：“旧金山的男性用户进行了多少次维基百科编辑？”此查询基于维度值的布尔表达式（city==’San Francisco’ and gender=’Male’）过滤表1中的维基百科数据集。在许多实际数据集中，维度列包含字符串，度量列包含数值。Druid为字符串列创建额外的查找索引，以便只扫描属于特定查询过滤器的那些行。
让我们考虑表1中的page列。对于表1中的每个唯一页面，可以使用一些标记来指明哪些行可以看到特定页面。我们可以将此信息存储在二进制数组中，其中数组索引表示我们的行。如果在特定行中看到特定页面，则该数组索引被标记为1.例如
Justin Bieber -> rows [0, 1] -> [1][1][0][0]
Ke$ha -> rows [2, 3] -> [0][0][1][1]
Justin Bieber在行0和1中可以看到。列值到行索引的映射形成了一个倒排索引[39]。要知道哪些行包含Justin Bieber或Ke$ha，我们可以对这两个数组进行OR运算。
[0][1][0][1] OR [1][0][1][0] = [1][1][1][1]
这种对大型bitmap数据集执行布尔运算的方法通常用于搜索引擎中。OLAP负载的位图索引在[32]中有详细描述。位图压缩算法是一个明确的研究领域[2，44，42]，并且经常利用游标编码算法。Druid选择使用Concise算法[10]。图7说明了使用整数数组进行Concise压缩的字节数。结果是在一个cc2.8xlarge系统生成的，其中使用了单线程，2G堆，512m年轻带，和每个运行之间的强制GC。数据集是从Twitter garden hose[41]数据流收集的一天的数据。数据集包含2,272,295行和12个不同基数的维度。作为一个额外的比较，我们也对数据集行排序以做到最大化压缩。

在未排序的情况下，Concise压缩后大小为53,451,144字节，总整数数组大小为127,248,520字节。总的来说，Concise压缩集比整数数组小42％。在排序的情况下，总Concise压缩大小为43,832,884字节，总的整数数组大小为127,248,520字节。有趣的是，在排序后，全局压缩只增加了很小一部分。

4.2 存储引擎

Druid的持久化组件允许不同的存储引擎以插件的方式接入，类似于Dynamo。这些存储引擎可以将数据存储在一个完全的in-memory结构的引擎中，例如JVM heap，或者是存储于 memory-mapped 结构的存储中。Druid中存储引擎可配置更换的这个能力依赖于一个特定的应用规范。一个in-memory的存储引擎要比memory-mapped存储引擎的成本昂贵得多，但是如果对于性能特别敏感的话，in-memory存储引擎则是更好的选择。默认情况下使用的是memory-mapped存储引擎。
当使用一个memory-mapped存储引擎的时候，Druid依赖于操作系统来对segment在内存中进行换入和换出操作。因为只有当segment加载到内存中了才可以被查询，所以memory-mapped存储引擎允许将最近的segment保留在内存中，而那些不会再被查询的segment则被换出。使用memory-mapped的主要缺点是当一个查询需要更多的segment并且已经超出了节点的内存容量时，在这种情况下，查询性能将会因为不断的在在内存中进行segment的换入和换出而下降。
来源:微信公众号:浪尖聊大数据
出处:
https://mp.weixin.qq.com/s?__biz=MzUyMDA4OTY3MQ==&mid=2247493929&idx=1&sn=04537c4ad49fd153b6f9936d6fd22ec1