Netflix：如何打造开放协作的实时 ETL 平台？

Netflix 致力于会员的喜悦。我们不懈地专注于提高产品体验和高质量内容。近年来，我们一直在技术驱动的 Studio 和内容制作方面进行大量投资。在这个过程中，我们发现在实时数据平台的领域里中出现了许多独特并有意思的挑战。例如，在微服务架构中，领域对象分布在不同的 App 及其有状态存储中，这使得低延迟高一致性的实时报告和实体搜索发现特别具有挑战性。

产品背景 

Netflix 的长久愿景是把欢乐和微笑带给整个世界，通过在全球各地拍摄一些高质量、多元化的内容产品放在平台上，分享给平台超过一个亿级别的用户。更远大的目标为了给用户带来愉悦的体验，Netflix 的努力方向分为两个：

1. 一方面是通过数据整合知识来反馈并用于提高用户的产品体验中去；

2. 另一方面通过建立一个技术驱动的 Studio 去帮助产出内容质量更高的产品。

而作为一个数据平台团队，需要关注的是怎么帮助公司中不同的开发人员、数据分析人员等实现其在公司中的价值，最终为解决上述两方面问题做出自己的贡献。

简单地介绍一下 Netflix 数据平台团队及相应的产品，Keystone。它的主要功能是帮助公司在所有的微服务中埋点、建立 Agent、发布事件、收集事件信息，然后放到不同的数据仓库中进行存储，比如 Hive 或 ElasticSearch，最后帮助用户在数据实时存储的情况下实现计算和分析。

• 从 用户 的角度来讲，Keystone 是一个完整的自容（Self-contained）的平台,支持多用户，用户可以通过所提供的 UI 很方便地声明并创建自己想要的 pipeline。

• 从 平台 角度来说，Keystone 提供底层所有分布式系统中实现比较困难的解决方案，如容器编排（Container Orchestration）、工作流管理（Workflow Management）等等，这些对于用户是不可见的。

• 从 产品 的角度来说，主要有两个功能，一个是帮助用户将数据从边缘设备移到数仓，另一个是帮助用户实时计算的功能。

• 从 数字 的角度来说， Keystone 在 Netflix 的使用是非常有必要的，只要跟数据打交道的开发者，一定会用到，因此  Keystone 在整个公司中有几千个用户，并有一百个 Kafka 的集群支持每天 10PB 数量级左右的数据。

Keystone 的整个架构分为两层，底层是 Kafka 和 Flink 作为底层引擎，底层对所有分布式系统中比较困难的技术方案进行抽象，对用户不可见，在上层构建整个应用；服务层会提供抽象的服务，UI 对于用户来讲比较简单，不需要关心底层实现。

下面介绍一下 Keystone 产品在过去四五年的发展历程。最初的动机是收集所有设备的数据并将其存储到数据仓库中，当时使用的是 Kafka 技术，因为数据移动比较好解决，本质上来讲仅是一个多并发的问题 。

在此之后，用户给出了新的需求，即在数据移动的过程中对数据进行一些简单的处理操作，比如筛选（Filter），还有一个很通用的功能 —— projection，为此 Keystone 推出了针对该需求推出了相应的功能特性。

经过一段时间后，用户表示想做更加复杂的 ETL，比如 Streaming Join 等，因此产品决定将底层的 API 提供给用户，并将底层的关于所有分布式系统的解决方案抽象化，让其更好地关注上层的内容。

产品功能 

产品功能介绍将围绕 Netflix 中的两个“超级英雄” Elliot 和 Charlie 来展开。Elliot 是来自数据科学工程组织的一个数据科学家，他的需求是在非常大的数据中寻找响应的 pattern，以帮助提高用户体验；Charlie 是一个来自 Studio 组织的应用开发者，其目标是通过开发一系列的应用来帮助周边的其他开发者产出更高质量的产品。

这两个人的工作对于产品来讲都非常重要，Elliot 的数据分析结果可以帮助给出更好的推荐和个性化定制,最终提高用户体验；而 Charlie 的工作可以帮助周边的开发者提高效率。

Recommendation & Personalization

Elliot 作为一个数据科学家，需要的是一个简单易用的实时 ETL 操作平台，他不希望写非常复杂的编码，同时需要保证整个 pipeline 的低延时。他所从事的工作和相关需求主要有以下几个：

• 推荐和个性化定制 。该工作中可以根据个人特点的不同将同样的视频通过不同的形式推送给相应的用户，视频可以分为多个 row，每一个 row 可以是不同的分类，根据个人的喜好可以对不同的 row 进行更改。此外，每一个视频的题目都会有一个 artwork，不同国家、不同地域的不同用户对 artwork 的喜好也可能不同，也会通过算法进行计算并定制适合用户的 artwork。

• A/B Testing 。Netflix 提供给非会员用户 28 天免费的视频观看机会，同时也相信给用户看到了适合自己的视频，用户更有可能会购买 Netflix 的服务，而在进行A/B Testing 的时候，就需要 28 天才能做完。对于 Elliot 来讲，进行 A/B Testing 的时候可能会犯错误，他所关心的是怎么样才能在不用等到 28 天结束的时候就可以提前发现问题。

当在设备上观看 Netflix 的时候，会以请求的形式和网关进行交互，然后网关会将这些请求分发给后端的微服务，比如说用户在设备上点击播放、暂停、快进、快退等操作，这些会有不同的微服务进行处理，因此需要将相应的数据收集起来，进一步处理。

对于 Keystone 平台团队来讲，需要收集不同的微服务中产生的数据并进行存储。Elliot 需要将不同的数据整合起来，以解决他关注的问题。

至于为什么要使用流处理，主要有四方面的考量，即实时报告、实时告警、机器学习模型的快速训练以及资源效率。相比于前两点，机器学习模型的快速训练以及资源效率对 Elliot 的工作更加重要。 尤其需要强调的是资源效率，针对前面的 28 天的 A/B Testing，目前的做法是每天将数据与前 27 天做 Batch Processing，这个过程中涉及了很多重复处理，使用流处理可以很好地帮助提高整体的效率。

Keystone 会提供命令行的 工具 给用户，用户只需要在命令行中输入相应的命令来进行操作，工具最开始会询问用户一些简单的问题，如需要使用什么 repository 等，用户给出相应的回答后，会最终产生一个模板，用户便可以开始使用工具进行开发工作；产品还提供一系列简单的 SDK，目前支持的是Hive、Iceberg、Kafka 和 ElasticSearch 等。

需要强调的是 Iceberg，它是在 Netflix 主导的一个 Table Format，未来计划取代 Hive。其提供了很多特色功能来帮助用户做优化； Keystone 向用户提供了简单的 API，可以帮助其直接生成 Source 和 Sink。

Elliot 在完成一系列的工作之后，可以选择将自己的代码提交到 repository 中，后台会自动启动一个 CI/CD pipeline，将所有的源代码和制品等包装在 Docker 镜像中，保证所有的版本一致性。Elliot 在 UI 处只需要选择想要部署哪一个版本，然后点击部署按钮可以将 jar 部署到生产环境中。

产品会在后台帮助其解决底层分布式系统比较困难的问题，比如怎么做容器编排等，目前是基于资源的编排，未来计划向 K8S 方向发展。部署 Job（作业）包的过程中会部署一个 JobManager 的集群和一个 TaskManager 的集群，因此每一个 Job 对于用户来说是完全独立的。

产品提供默认的配置选项，同时也支持用户在平台 UI 上修改并覆盖配置信息，直接选择部署即可生效，而不需重写代码。Elliot 之前有一个需求是在 Stream Processing 的过程中，比如从不同的 Topic 中去读取数据，出现问题的情况下可能需要在 Kafka 中操作，也可能需要在数据仓库中操作，面对该问题，其需求是在不改动代码的情况下切换不同的 Source，而目前平台提供的UI很方便地完成该需求。此外平台还可以帮助用户在部署的时候选择需要多少资源来运行作业。

很多用户从 Batch Processing 转到 Stream Processing 的过程中，已经有了很多需要的制品，比如 Schema 等，因此平台还帮助其简单地实现这些制品的集成。

平台拥有很多需要在其之上写 ETL 工程的用户，当用户越来越多的时候，平台的可伸缩性显得尤为重要。为此，平台采用了一系列的 pattern 来解决该问题。具体来讲，主要有三个 pattern 正在使用，即 Extractor Pattern、Join Pattern 和 Enrichment Pattern。

Content Production  

先简要介绍一下什么是 Content Production。包括预测在视频制作方面的花费、制定 program、达成 deal、制作视频、视频后期处理、发布视频以及金融报告。

Charlie 所在的是 Studio 部门主要负责开发一系列的应用来帮助支持 Content Production。每一个应用都是基于微服务架构来开发部署的，每一个微服务应用会有自己的职责。举个最简单的例子，会有专门管理电影标题的微服务应用，也会有专门管理 deals 和 contracts 的微服务应用等等。

面对如此多的微服务应用，Charlie 面临的挑战问题是当其在进行实时搜索的过程中，比如搜索某一个电影的演员，需要将数据从不同的地方 join 起来；另外数据每天都在增加，保证实时更新的数据的一致性比较困难，这本质上是分布式微服务系统的特点导致，不同的微服务选择使用的数据库可能不同，这给数据一致性的保证又增加了一定的复杂度。针对该问题，常用的解决方案有以下三个：

• Dual writes: 当开发者知道数据需要放到主要的数据库中的时候，同时也要放到另一个数据库中，可以很简单地选择分两次写入到数据库中，但是这种操作是不容错的，一旦发生错误，很有可能会导致数据的不一致；

• Change Data Table:  需要数据库支持事务的概念，不管对数据库做什么操作，相应的变更会加到事务变更的 statement 中并存入单独的表中，之后可以查询该 change 表并获取相应的变更情况并同步到其他数据表；

• Distributed Transaction： 指的是分布式事务，在多数据环境中实现起来比较复杂。

Charlie 的一个需求是将所有的电影从 Movie Datastore 复制到一个以 Elasticsearch 来支持的 movie search index 中，主要通过一个 Polling System 来做数据拉取和复制，数据一致性的保证采用的是上述的 Change data table 的方法。

该方案的弊端是只支持定期数据拉取，另外 Polling System 和数据源直接紧密结合，一旦 Movie Search Datastore 的 Schema 改变，Polling System 就需要修改。为此，该架构在后来做了一次改进，引入了事件驱动的机制，读取数据库中所有实现的事务，通过 stream processing 的方式传递到下一个 job 进行处理。为了普适化该解决方案，在 source 端实现了不同数据库的 CDC（Change Data Capture）支持，包括 MySQL 、PostgreSQL 和 Cassandra 等在 Netflix 中比较常用的数据库，通过 Keystone 的 pipeline 进行处理。

挑战及解决方案

下面分享一下上述方案存在的挑战和相应的解决方案：

• Ordering Semantics

在变更数据事件中，必须要保证 Event ordering，比如一个事件包含 create、update 和 delete 是三个操作，需要返回给消费者侧一个严格遵守该顺序的操作事件。一个解决方案是通过 Kafka 来控制；另一个解决方案是在分布式系统中保证捕获的事件与实际从数据库中读取数据的顺序是一致的，该方案中当所有的变更事件捕获出来后，会存在重复和乱序的情况，会通过 Flink 进行去重和重新排序。

• Processing Contracts

在写 stream processing 的时候，很多情况下不知道 Schema 的具体信息，因此需要在消息上定义一个契约 contract，包括 Wire Format 以及在不同的层级上定义与Schema 相关的信息，如基础设施（Infrastructure）、平台（Platform）等。Processor Contract 的目的是帮助用户将不同的 processor metadata 组合起来，尽量减少其写重复代码的可能。

举一个具体的案例，比如 Charlie 希望有新的 deal 的时候被及时通知，平台通过将相关的不同组件组合起来，DB Connector、Filter 等，通过用户定义契约的方式帮助其实现一个开放的可组合的流数据平台。

以往所看到的 ETL 工程大多数适用于数据工程师或数据科学家。但从经验上来讲，ETL 的整个过程，即 Extract、Transform 和 Load，其实是有被更广泛应用的可能。最早的 Keystone 简单易用，但灵活性不高，之后的发展过程中虽然提高了灵活性，但复杂性也相应地增大了。因此未来团队计划在目前的基础上进一步优化，推出开放的、合作的、可组合的、可配置的 ETL 工程平台，帮助用户在极短的时间解决问题。

作者简介：

徐振中，Netflix 软件工程师，在 Netflix 从事高度可扩展和弹性的流媒体数据平台的基础设施工作，热衷于研究分享与实时数据系统、分布式系统基本原理相关的任何有趣的事情！

点击「 阅读原文 」可查看作者现场分享视频～

关注 Flink 中文社区，获取更多技术干货

你也「 在看 」吗？ :point_down: