流媒体与实时计算，Netflix公司Druid应用实践

Netflix(Nasdaq NFLX)，也就是网飞公司，成立于1997年，是一家在线影片[租赁]提供商，主要提供Netflix超大数量的[DVD]并免费递送，总部位于美国加利福尼亚州洛斯盖图。1999年开始订阅服务。2009年，该公司可提供多达10万部DVD电影，并有1千万的订户。2007年2月25日，Netflix宣布已经售出第10亿份DVD。

Netflix已经连续五次被评为顾客最满意的网站。能够经过PC、TV及iPad、iPhone收看电影、电视节目，可经过[Wii]，[Xbox360]，[PS3]等设备链接TV。Netflix是全球领先的经营在线业务公司。它成功地把传统的影像租赁业务和现代化的市场营销手段、先进的IT网络技术结合起来，从而开创了在线影像租赁的新局面。Netflix经过整合其自身的营销手段和最近的IT网络技术，成功地改变了消费习惯和打造了本身的品牌优点。

Netflix在2011年开始探索自制内容的举动并不被看好。直到2013年，其首部自制剧《纸牌屋》取得爆红后，舆论冲击及股票下滑的趋势才得以扭转。这也让Netflix成功打响了平台自制内容的第一炮。

2019 年 7 月 4 日，网飞的原创剧《怪奇物语》第三季开播，一如往常地一口气放出 12 集，再次掀起话题热潮。取得这样的成功，网飞天然是高兴不已。7 月 8 日，这家一般并不爱自诩成绩的公司表示，有近 4100 万家庭在四天以内观看了《怪奇物语》最新季，超过 1800 万家庭已经把整 8 集所有刷完。若是须要对比数据的话，4 月份 HBO 发布的《权力的游戏》最终季首播集观看人数为 1740 万。

在持续推进创新技术更新的同时，Netflix确保始终如一的出色的用户体验绝非易事。如何才能确信更新系统的时候不会影响用户的使用？并且实际上如何获得更多的反馈，能够对系统进行不断地改进也是一个巨大的挑战。

最终，Netflix公司经过对设备的数据进行采集，使用来自设备的实时日志做为事件源，获得了大量的数据，经过实时的大数据了解和量化了用户设备，最终成功的近乎无缝地处理了视频的浏览和回放，完美的解决了这些问题。

下面咱们来具体了解一下：

系统架构

如上图，整个系统架构经过对用户设备日志收集，经过kafka的消息传递，最终存储在Druid中。

一旦有了这些数据，就将它们存入数据库，这里使用的是实时分析数据库Druid。

每项数据流均标有关于所用设备类型的匿名详细信息，例如，该设备是智能电视，iPad仍是Android手机。这使得可以根据各个方面对设备进行分类并查看数据。反过来，这又使咱们可以定向的分析仅影响特定人群的问题，例如应用程序的版本，特定类型的设备或特定国家/地区。

可经过仪表板或临时查询当即使用此聚合数据进行查询。还能够连续检查指标是否有警报信号，例如新版本是否正在影响某些用户或设备的播放或浏览。这些检查用于警告负责的团队，他们能够尽快解决该问题。

在软件更新期间，为部分用户启用新版本，并使用这些实时指标来比较新版本与之前版本的性能。指标中的任何问题都会使咱们马上发现并停止更新，并将那些使新版本直接恢复到先前版本。

因为每秒须要处理超过200万个事件，所以将其放入能够快速查询的数据库是一个很是艰巨的任务。咱们须要一个拥有足够的性能与多维度查询的数据库，来处理天天产生超过1,150亿行的数据。在Netflix，最终选择利用Apache Druid来应对这一挑战。

Druid（德鲁伊）

Druid是一个分布式的支持实时分析的数据存储系统。通俗一点：高性能实时分析数据库。

Apache Druid是一个高性能的实时分析数据库。它是为须要快速查询和提取的工做流而设计的。德鲁伊在即时数据可视性，即席查询，运营分析和处理高并发方面表现出色。” — druid.io

所以，Druid很是适合如今咱们面临的这种用例。事件数据的摄取频率很是高，具备大数据量和快速查询要求。

Druid不是关系数据库，可是某些概念是可移植的。咱们有数据源，而不是表。与关系数据库同样，这些是表示为列的数据的逻辑分组。Druid的Join性能目前还不是很优秀。所以，咱们须要确保每一个数据源中都包含咱们要过滤或分组依据的任何列。

数据源中主要有三类列-时间，维度和指标。

德鲁伊中的一切都取决于时间。每一个数据源都有一个timestamp列，它是主要的分区机制。维度是可用于过滤，查询或分组依据的值。指标是能够汇总的值，几乎老是数字。

咱们假设数据由时间戳做为键，Druid能够对存储，分配和查询数据的方式进行一些优化，从而使咱们可以将数据源扩展到数万亿行，而且仍然能够实现查询响应时间在十毫秒内。

为了达到这种级别的可伸缩性，Druid将存储的数据划分为多个时间块。时间块的持续时间是可配置的。能够根据您的数据和用例选择适当的持续时间。对于咱们的数据和用例，咱们使用1小时时间块。时间块内的数据存储在一个或多个段中。每一个段都保存有全部数据行，这些行均落在其时间戳键列所肯定的时间块内。能够配置段的大小，以使行数或段文件的总大小有上限。

查询数据时，Druid将查询发送到集群中全部包含查询范围内时间块的分段的节点。每一个节点在将中间结果发送回查询代理节点以前，都会对所保存的数据进行并行处理。代理将执行最终合并和聚合，而后再将结果集发送回客户端。

摄取数据

把数据实时插入到此数据库。这些事件（在本例中为指标）不是从单个记录插入到数据源中，而是从Kafka流中读取。每一个数据源使用1个主题。在Druid中，咱们使用Kafka索引编制任务，该任务建立了多个在实时节点中间管理者之间分布的索引编制工做器。

这些索引器中的每个都订阅该主题，并从流中读取其事件共享。索引器根据摄入规范从事件消息中提取值，并将建立的行累积在内存中。一旦建立了行，就能够对其进行查询。到达索引器仍在填充一个段的时间块的查询将由索引器自己提供。因为索引编制任务实际上执行两项工做，即摄取和现场查询，所以及时将数据发送到“历史节点”以更优化的方式将查询工做分担给历史节点很是重要。

Druid能够在提取数据时对其进行汇总，以最大程度地减小须要存储的原始数据量。汇老是一种汇总或预聚合的形式。在某些状况下，汇总数据能够极大地减小须要存储的数据大小，从而有可能将行数减小几个数量级。可是，减小存储量确实要付出必定的代价：咱们失去了查询单个事件的能力，只能以预约义的查询粒度进行查询。对于咱们的用例，咱们选择了1分钟的查询粒度。

在提取期间，若是任何行具备相同的维度，而且它们的时间戳在同一分钟内（咱们的查询粒度），则这些行将被汇总。这意味着经过将全部度量值加在一块儿并增长一个计数器来合并行，所以咱们知道有多少事件促成了该行的值。这种汇总形式能够显着减小数据库中的行数，从而加快查询速度，由于这样咱们就能够减小要操做和聚合的行。

一旦累积的行数达到某个阈值，或者该段已打开太长时间，则将这些行写入段文件中并卸载到深度存储中。而后，索引器通知协调器段已准备好，以便协调器能够告诉一个或多个历史节点加载该段。一旦将段成功加载到“历史”节点中，就能够从索引器中将其卸载，而且历史记录节点如今将为全部针对该数据的查询提供服务。

数据管理

就像您想象的那样，随着维数基数的增长，在同一分钟内发生相同事件的可能性下降。管理基数以及所以汇总，是得到良好查询性能的有力手段。

为了达到所需的摄取速率，咱们运行了许多索引器实例。即便在索引任务中合并了相同行的汇总，在相同的索引任务实例中得到全部相同行的机会也很是低。为了解决这个问题并实现最佳的汇总，咱们安排了一个任务，在将给定时间块的全部段都移交给历史节点以后运行。

计划的压缩任务从深度存储中获取全部分段以进行时间块化，并执行映射/缩小做业以从新建立分段并实现完美的汇总。而后，由“历史记录”节点加载并发布新的细分，以替换并取代原始的，较少汇总的细分。在咱们的案例中，经过使用此额外的压缩任务，咱们发现行数提升了2倍。

知道什么时候收到给定时间块的全部事件并非一件容易的事。可能有关于Kafka主题的迟到数据，或者索引器可能会花一些时间将这些片断移交给“历史”节点。为了解决此问题，咱们在运行压缩以前强加了一些限制并执行检查。

首先，咱们丢弃任何很是迟到的数据。咱们认为这太旧了，没法在咱们的实时系统中使用。这样就能够肯定数据的延迟时间。其次，压缩任务是有延迟地安排的，这给了段足够的时间以正常流程分流到历史节点。最后，当给定时间块的计划压缩任务开始时，它查询段元数据以检查是否还有任何相关段仍在写入或移交。若是有，它将等待几分钟后重试。这样能够确保全部数据都由压缩做业处理。

若是没有这些措施，咱们发现有时会丢失数据。开始压缩时仍要写入的段将被具备更高版本的新压缩的段覆盖，所以具备优先权。这有效地删除了还没有完成移交的那些段中包含的数据。

查询方式

Druid支持两种查询语言：Druid SQL和原生查询。在后台，Druid SQL查询被转换为本地查询。原生查询做为JSON提交到REST端点，这是咱们使用的主要机制。

对集群的大多数查询都是由自定义内部工具（例如仪表板和警报系统）生成的。这些系统最初旨在与咱们内部开发的开源时间序列数据库Atlas一块儿使用。所以，这些工具使用Atlas Stack查询语言。

为了加快采用Druid查询的速度并实现对现有工具的重用，咱们添加了一个转换层，该层接受Atlas查询，将其重写为Druid查询，发布查询并将结果从新格式化为Atlas结果。这个抽象层使现有工具能够按原样使用，而且不会为用户访问咱们的Druid数据存储中的数据建立任何额外的学习曲线。

调整

在调整群集节点的配置时，咱们以很高的速度运行了一系列可重复和可预测的查询，以便得到每一个给定配置的响应时间和查询吞吐量的基准。这些查询旨在隔离集群的各个部分，以检查查询性能是否有所改善或下降。

例如，咱们针对最新数据运行了有针对性的查询。一样，对于更长的持续时间，但只有较旧的数据能够确保咱们仅查询“历史”节点以测试缓存配置。再次使用按很是高的基数维度分组的查询，以检查结果合并是如何受到影响的。咱们继续调整并运行这些基准测试，直到对查询性能感到满意为止。

在这些测试中，咱们发现调整缓冲区的大小，线程数，查询队列长度和分配给查询缓存的内存对查询性能产生了有效影响。可是，引入压缩工做将占用咱们汇总不良的细分，并以完美汇总将它们从新压缩，这对查询性能产生了更大的影响。

咱们还发现，在历史节点上启用缓存很是有好处，而在代理节点上启用缓存则没有那么多。太多了，咱们不使用代理上的缓存。这多是因为咱们的用例所致，可是咱们几乎进行的每一个查询都未命中代理上的缓存，这多是由于查询一般包含最新数据，由于这些数据始终会到达，所以不会包含在任何缓存中。

摘要

通过屡次迭代，针对咱们的用例和数据速率进行了调整和定制，德鲁伊已被证实具备咱们最初但愿的能力。

咱们已经可以使用功能强大且可用的系统，可是还有更多工做要作。咱们的摄入量和摄入率不断提升，查询的数量和复杂性也在不断增长。随着更多团队实现这些详细数据的价值，咱们常常添加更多指标和维度，从而推进系统更加努力地工做。咱们必须继续监视和调整，以保持查询性能。

目前，咱们每秒接收超过200万个事件，并查询超过1.5万亿行，以深刻了解咱们的用户如何体验该服务。全部这些都有助于咱们保持高质量的Netflix体验，同时实现不断的创新。

实时流式计算与流媒体的碰撞才刚刚开始，而Druid做为一款极易上手的高性能实时查询数据库，也会获得愈来愈多的普遍使用。

更多实时数据分析相关博文与科技资讯，欢迎关注 “实时流式计算”

获取《Druid实时大数据分析》电子书，请在公号后台回复 “Druid”