Netflix如何使用Druid以确保高质量的体验

在不断推动创新技术更新的同时，确保Netflix始终如一的出色体验绝非易事。我们如何才能确信更新不会伤害我们的用户？当我们打算做的时候，我们真的在做可衡量的改进？

使用回放设备的实时日志作为事件源，我们得出测量结果，以便理解和量化用户的设备如何无缝地处理浏览和回放。

一旦我们有了这些措施，我们就把它们输入数据库。每一项测量都会标记出使用哪种设备的匿名细节，例如，该设备是智能电视、iPad还是Android手机。这使我们能够根据各个方面对设备进行分类和查看数据。这反过来又允许我们隔离可能只影响某一组的问题，例如某一版本的应用程序、某些类型的设备或特定的国家/地区。

此聚合数据可立即用于查询，可以通过仪表板或特殊查询进行查询。这些指标还不断地检查报警信号，例如新版本是否影响某些用户或设备的播放或浏览。这些检查用于提醒责任团队尽快解决问题。

在软件更新期间，我们为一部分用户启用新版本，并使用这些实时度量来比较新版本与以前版本的性能。度量中的任何回归都会给我们一个信号，让我们中止更新，并将那些获得新版本的用户恢复到以前的版本。

随着这些数据以每秒超过200万个事件的速度到达，将其放入一个可以快速查询的数据库是非常困难的。我们需要足够的数据维度来帮助隔离问题，因此我们每天生成超过1150亿行。在Netflix，我们利用ApacheDruid来帮助以我们的规模应对这一挑战。

Druid

Apache Druid是一个高性能的实时分析数据库。它是为那些快速查询和接收非常重要的工作流而设计的。Druid擅长于即时数据可见性、即席查询、操作分析和处理高并发性

因此，Druid非常适合我们的用例。事件数据摄取率高，基数高，查询速度快。

Druid不是一个关系数据库，但是一些概念是可以转移的。我们有数据源，而不是表。与关系数据库一样，它们是以列表示的数据的逻辑分组。与关系数据库不同，没有连接的概念。因此，我们需要确保每个数据源中都包含要筛选或分组的列。

数据源中主要有三类列：时间、维度和度量。

Druid的一切都是由时间决定的。每个数据源都有一个timestamp列，它是主要的分区机制。维度是可用于筛选、查询或分组的值。度量是可以聚合的值，几乎总是数字。

通过取消执行连接的功能，并假设数据是由时间戳设置的，Druid可以对其存储、分发和查询数据的方式进行一些优化，这样我们就能够将数据源扩展到数万亿行，并且仍然可以在10毫秒内实现查询响应时间。

为了达到这个级别的可伸缩性，Druid将存储的数据划分为时间块。时间块的持续时间是可配置的。可以根据您的数据和用例选择适当的持续时间。对于我们的数据和用例，我们使用1小时的时间块。时间块中的数据存储在一个或多个段中。每个段包含所有在时间块（由其timestamp键列确定）内的数据行。可以配置段的大小，使行数或段文件的总大小有一个上限。

在查询数据时，Druid将查询发送到集群中的所有节点，这些节点在查询范围内保存时间块的片段。在将中间结果发送回querybroker节点之前，每个节点在它所持有的数据上并行处理查询。在将结果集发送回客户端之前，代理将执行最终的合并和聚合。

Ingestion

实时插入此数据库。事件（在本例中是度量）是从Kafka流中读取的，而不是将单个记录插入到数据源中。我们每个数据源使用一个主题。在Druid中，我们使用Kafka索引任务来创建多个索引工作者，这些索引工作者分布在实时节点（中间管理器）中。

每个索引器都订阅主题并从流中读取其事件共享。索引器根据摄取规范从事件消息中提取值，并在内存中累积创建的行。一旦创建了一行，就可以查询它。到达某个时间块的查询（其中某个段仍由索引器填充）将由索引器自己提供服务。由于索引任务基本上是执行两个作业，即摄取和部署查询，因此必须及时将数据输出到历史节点，以便以更优化的方式将查询工作卸载给它们。

Druid可以在接收数据时将其卷起，以最小化需要存储的原始数据量。汇总是一种汇总或预聚合的形式。在某些情况下，汇总数据可以极大地减少需要存储的数据的大小，从而有可能按数量级减少行数。然而，这种存储减少确实是有代价的：我们失去了查询单个事件的能力，只能以预定义的查询粒度进行查询。对于我们的用例，我们选择了1分钟的查询粒度。

在接收过程中，如果任何行具有相同的维度，并且它们的时间戳在同一分钟内（我们的查询粒度），那么这些行将上卷。这意味着通过将所有度量值相加并增加一个计数器来组合这些行，这样我们就知道有多少事件贡献了这一行的值。这种形式的汇总可以显著减少数据库中的行数，从而加快查询速度，因为这样可以减少要操作和聚合的行数。

一旦累积的行数达到某个阈值，或者段已打开太长时间，这些行将写入段文件并卸载到深度存储。然后索引器通知协调器该段已准备就绪，以便协调器可以告诉一个或多个历史节点加载它。一旦该段成功加载到历史节点中，它就会从索引器中卸载，任何针对该数据的查询现在都将由历史节点提供服务。

数据管理

正如你所想象的，随着维度基数的增加，在同一分钟内发生相同事件的可能性降低。管理基数（因此是上卷）是实现良好查询性能的强大杠杆。

为了达到我们需要的摄取率，我们运行了许多索引器实例。即使汇总将索引任务中的相同行组合在一起，在索引任务的同一实例中获得这些相同行的可能性也非常低。为了解决这一问题并实现最佳的汇总，我们安排一个任务在给定时间块的所有段都交给历史节点之后运行。

此计划的压缩任务从深度存储中获取时间块的所有片段，并运行map/reduce作业来重新创建片段并实现完美的汇总。然后，新的段由历史节点加载和发布，以替换和替换原来的卷取较少的段。在我们的例子中，我们看到通过使用这个额外的压缩任务，行数提高了2倍。

知道某个给定时间段的所有事件何时都已收到并非易事。卡夫卡主题的数据可能会延迟到达，或者索引器可能会花时间将片段交给历史节点。为了解决这个问题，我们在运行压缩之前执行一些限制和检查。

首先，我们丢弃任何非常晚到达的数据。我们认为它太老了，在我们的实时系统中没有用处。这对数据的延迟时间设置了一个界限。其次，压缩任务被延迟调度，这使得在正常流中有足够的时间将数据段卸载到历史节点。最后，当给定时间块的计划压缩任务开始时，它查询段元数据以检查是否有任何相关的段仍在写入或传递。如果有，它将等待几分钟后重试。这样可以确保压缩作业处理所有数据。

如果没有这些措施，我们发现有时我们会丢失数据。压缩开始时仍在写入的段将被具有更高版本的新压缩段覆盖，因此优先。这实际上删除了那些尚未完成移交的数据段中包含的数据。

查询

Druid支持两种查询语言：druidsql和原生查询。在这种情况下，druidsql查询被转换为本机查询。本机查询作为JSON提交给REST端点，这是我们使用的主要机制。

对集群的大多数查询都是由定制的内部工具（如仪表板和警报系统）生成的。这些系统最初设计用于我们内部开发的、开源的时间序列数据库Atlas。因此，这些工具使用Atlas堆栈查询语言。

为了加快查询Druid的采用，并支持现有工具的重用，我们添加了一个转换层，它接受Atlas查询，将它们重写为Druid查询，发出查询并将结果重新格式化为Atlas结果。这个抽象层使现有的工具能够按原样使用，并且不会为用户访问我们的Druid数据存储中的数据创建额外的学习曲线。

调整比例

在调整集群节点配置的同时，我们以高速率运行一系列可重复和可预测的查询，以获得每个给定配置的响应时间和查询吞吐量的基准。这些查询被设计用来隔离集群的各个部分，以检查查询性能的改进或退化。

例如，我们对最新的数据运行有针对性的查询，以便只查询中层管理人员。同样地，对于更长的持续时间，但只有较旧的数据，以确保我们只查询历史节点来测试缓存配置。再次使用按非常高的基数维度分组的查询来检查结果合并是如何受到影响的。我们继续调整和运行这些基准测试，直到我们对查询性能满意为止。

在这些测试中，我们发现调整缓冲区大小、线程数、查询队列长度和分配给查询缓存的内存对查询性能有有效的影响。但是，压缩作业的引入对查询性能有更大的影响，该作业将使用我们的卷取较差的片段，并使用完美的卷取对它们进行重新压缩。

我们还发现，在历史节点上启用缓存是非常有益的，而在代理节点上启用缓存则要少得多。以至于我们不在经纪人身上使用缓存。这可能是由于我们的用例造成的，但几乎我们所做的每个查询都会丢失代理上的缓存，可能是因为查询通常包含最新的数据，而这些数据不会在任何缓存中，因为它们总是到达。

总结

在为我们的用例和数据速率进行了多次反复的调优和裁剪之后，Druid已经证明了我们最初希望的能力。

我们已经能够得到一个有能力和可用的系统，但还有更多的工作要做。我们的摄取量和速率在不断增加，查询的数量和复杂性也在不断增加。随着更多团队意识到这些详细数据的价值，我们经常添加更多的度量和维度，从而推动系统更加努力地工作。我们必须继续监视和调优以保持查询性能良好。

我们目前每秒接收超过200万个事件，并查询超过1.5万亿行，以获得用户体验服务的详细信息。所有这些都有助于我们保持高质量的Netflix体验，同时实现不断创新。

原文链接：https://netflixtechblog.com/how-netflix-uses-druid-for-real-time-insights-to-ensure-a-high-quality-experience-19e1e8568d06