数据流动方式迭代：Netflix Studio 的 Data Mesh 实践

深度学习与Python

发布于 2023-04-01 16:31:25

1.1K0

发布于 2023-04-01 16:31:25

文章被收录于专栏：深度学习与python

作者 | Netflix 技术博客

译者 | 刘雅梦

策划 | 蔡芳芳

借助最新的数据网格平台（Data Mesh Platform），Netflix Studio 中的数据移动进入到了一个新阶段。这种配置驱动的平台在创建新管道时显著地缩短了前置时间，同时提供了新的支持特性，比如端到端的模式演进（schema evolution）、自助式 UI 和安全数据访问等。

1背景

未来几年，Netflix 上的大部分内容都将来自其自己的工作室（Netflix Studio）。Netflix 电影或电视据从开始宣传到在 Netflix 上映，需要经历许多阶段。这种规模是前所未有的，并且带来了许多有趣的挑战；其中一个挑战是如何跨多个阶段和系统提供 Studio 数据的可视化，以促进运营的卓越性并增强决策能力。Netflix 以其松耦合的微服务架构和全球工作室而闻名，使得从微服务到工作室数据目录的实时数据呈现及连接变得比以往任何时候都重要。

运营报告（Operational Reporting）是一种专门用于覆盖高分辨率、低延迟数据集的报告范式，为业务领域的详细日常活动和流程提供服务。这种范式旨在通过特定分析、决策支持和跟踪（任务、资产、进度等）等方式，帮助一线运营人员和利益相关方执行他们的任务。该范式跨越了方法、工具和技术，通常它的定义与分析报告（Analytical Reporting ）和预测建模（Predictive Modeling）的定义相反，后者在本质上更具有战略性（vs 战术性）。

在 Netflix Studio，团队构建了各种业务数据视图，为日常决策提供可视化。借助可靠的近实时数据，Studio 团队能够更好地跟踪和应对不断变化的制作节奏，并使用最新的信息来提高全球业务运营的效率。整个 Netflix Studio 之间的数据连接和 Operational Reporting 工具的可用性也避免了 Studio 用户形成数据孤岛。

2旅程

在过去的几年里，Netflix Studio 经历了几次数据移动方式的迭代。在初始阶段，数据消费者通过建立 ETL 管道，直接从数据库中提取数据。通过这种批处理的方式，出了一些问题，如数据移动是与数据库的表紧密耦合的，数据库模式不是业务数据模型的精确映射，数据陈旧（因为它不是实时的）等等。随后，我们转向了事件驱动的流数据管道（由 Delta 提供支持），与批处理的方式相比，它解决了一些问题，但也有其自身的痛点，比如流处理技术的高学习曲线、手动管道设置、缺乏模式演进（schema evolution）支持、新实体的加入效率低下、安全访问模型不一致等。

借助最新的数据网格平台（Data Mesh Platform），Netflix Studio 中的数据移动进入到了一个新阶段。这种配置驱动的平台在创建新管道时显著地缩短了前置时间，同时提供了新的支持特性，比如端到端的模式演进（schema evolution）、自助式 UI 和安全数据访问等。下面的图表显示了用于 Operational Reporting 数据移动的最新架构。

Operational Reporting 架构概述

对于数据传递，我们利用 Data Mesh 平台来推动数据移动。Netflix Studio 应用程序通过 Studio Edge 发布 GraphQL 查询，后者是一个连接 Netflix Studio 中的所有数据并提供一致性数据检索的统一 Graph。变更数据捕获（Change Data Capture，CDC）源连接器从 Studio 应用程序的数据库事务日志中读取并发出变更事件。CDC 事件被传递到 Data Mesh 扩展处理器中，该处理器向 Studio Edge 发出 GraphQL 查询以扩充数据。一旦数据落在 Netflix 数据仓库的 Iceberg 表中，它们就可以用于临时或预定的查询及报告。集中化的数据将被转移到第三方服务中，如为利益相关方提供的 Google Sheets 和 Airtable。我们将在下面几节中深入探讨数据传递（Data Delivery）和数据消费（Data Consumption）。

3通过 Data Mesh 进行数据传递

什么是 Data Mesh？

Data Mesh（数据网格）是一种完全托管的流式数据管道产品，用于支持变更数据捕获（CDC）用例。在 Data Mesh 中，用户创建源（source）并构建管道。源模拟外部管理源的状态——当外部源发生变更时，会在 Data Mesh 源中生成相应的 CDC 消息。可以将管道配置成转换并存储数据到外部托管的接收器（sink）中。

Data Mesh 提供了一个拖放式的自助服务用户界面，用于探测源和创建管道，这样用户就可以专注于交付业务价值，而无需担心管理和扩展复杂的数据流基础设施。

CDC 及数据源

变更数据捕获（Change data capture，CDC）是一种语义，用于处理源中的变更，以便将这些变更复制到接收器（sink）中。表变更可以是行变更（插入行、更新行、删除行）或模式（schema）变更（添加列、更改列、删除列）。到目前为止，CDC 数据源已经在 Netflix（MySQL，Postgres）的数据存储中实现。CDC 事件也可以通过 Java 客户端生产者库发送到 Data Mesh。

可重用的处理器及配置驱动

在 Data Mesh 中，处理器是一个可配置的数据处理应用程序，用于消费、转换和生成 CDC 事件。处理器有 1 个或多个输入以及 0 个或多个输出。具有 0 个输出的处理器是 sink 连接器；将事件写入外部托管的接收器中（例如 Iceberg、ElasticSearch 等）。

具有不同输入 / 输出的处理器

Data Mesh 允许开发人员为平台贡献处理器。处理器不一定是集中开发和管理的。但是，Data Mesh 平台团队致力于提供和管理利用率最高的处理器（比如 source 连接器和 sink 连接器）

处理器是可重用的。对于处理器的所有实例，将多次使用同一个处理器的镜像包。每个实例都相互适配。比如，可以配置一个 GraphQL 丰富处理器来查询 GraphQL 服务，以丰富不同管道中的数据；Iceberg sink 处理器可以多次初始化，以将数据写入到具有不同模式的不同数据库 / 表中。

端到端的模式演进

模式（Schema）是 Data Mesh 的关键组件。当上游模式发生变更时（例如 MySQL 表中的模式变更），Data Mesh 会检测到该变更，检查兼容性并将该变更应用到下游。通过模式演进（Schema Evolution），Data Mesh 可以确保 Operational Reporting 管道能始终使用最新的模式来生成数据。

我们将介绍 Data Mesh Schema 模式领域的几个核心概念。

消费者模式

消费者模式（Consumer Schema）定义了下游处理器如何使用数据的方式。请参见下面的示例。

消费者模式示例

模式兼容性

Data Mesh 使用消费者模式（Consumer Schema）的兼容性来实现灵活而安全的模式演进。如果 Operational Reporting 管道使用的字段已从 CDC 源中删除了，Data Mesh 会将此变更归类为 不兼容（incompatible），暂停管道处理并通知管道所有者。另一方面，如果某个必需字段并未被任何消费者使用，那么删除此类字段也是 兼容（compatible） 的。

两种类型的处理器

将所有字段从上游一直传递到下游

示例：过滤器处理器（Filter Processor）、接收器处理器（Sink Processors）

选择加入模演进示例

使用上游字段的子集。

示例：项目处理器，丰富处理器

选择退出模式演进示例

在 Data Mesh 中，我们引入了 “Opt-in to Schema Evolution” 的布尔标志来区分这两种类型的用例。

选择加入（Opt in）：所有上游字段都将被传播到处理器中。例如，当上游添加一个新字段时，它将自动传播。
选择退出（Opt out）：只有一部分字段（使用 “Is Consumed” 复选框来定义）会在处理器中传播并使用。其余字段的上游变更不会影响到该处理器。

模式传播

检查完模式的兼容性之后，Data Mesh 平台将根据最终用户的意图传播模式变更。使用“opt-in to schema Evolution”标志，Operational Reporting 管道可以使模式与上游数据存储保持同步。作为模式传播的一部分，平台还将模式从管道同步到 Iceberg sink 中。

模式演进图

通过 GraphQL 丰富处理器

在当前的 Data Mesh Operational Reporting 管道中，最常用的中间处理器是 GraphQL 丰富处理器（GraphQL Enrichment Processor）。它将来自源接收器（Source Connector）的 CDC 事件的列值作为 GraphQL 查询输入，然后向 Studio Edge 提交查询以丰富数据。借助 Studio Edge 的单一数据模型，它聚合了数据建模的工作，Studio UI 应用程序、后端服务和搜索平台都高度利用了这些工作。通过 Studio Edge 来丰富数据可以帮助我们在 Operational Reporting 的整个生态系统中实现一致的数据建模。

下面是一个 GraphQL 处理器配置的示例，管道构建器只需要配置以下字段就可以提供丰富处理器：

GraphQL 丰富处理器配置示例

下图是生产环境中用于接收电影相关数据的一个 Operational Reporting 管道示例。想要移动数据的团队不再需要学习和编写自定义的流处理作业。相反，他们只需要在 UI 中配置管道拓扑，同时能获得其他开箱即用的特性，如模式演进和安全数据访问等。

Operational Reporting 管道示例

4Iceberg Sink

Apache Iceberg 是一种用于大型分析数据集的开源表格格式。Data Mesh 利用 Iceberg 表来作为下游分析用例的数据仓库 sink 接收器。目前仅添加了 Iceberg sink。视图建立在原始 Iceberg 表的上面，以根据操作时间戳来检索每个主键的最新记录，该操作时间戳表明了记录何时在 sink 中生成。当前的管道消费者直接使用视图而不是原始表。

为了优化业务视图上下游查询的性能，以及降低 S3 GET OBJECT 操作的成本，需要一个压缩进程。一个每日运行一次的进程会按时间戳对记录进行排序，以生成压缩记录的数据帧。旧数据文件会被一组只包含压缩数据的新数据文件覆盖。

数据质量

Data Mesh 在处理器和管道级别上为操作的可观察性提供了度量指标和仪表板。如果管道出现问题，Operational Reporting 管道的所有者都将收到报警。我们还对 Data Mesh 管道生成的数据表进行了两种类型（端到端审计、人工综合事件审计）的审计，以保证数据质量。

在 Iceberg 表上创建的大多数业务视图可以容忍几分钟的延迟。然而，最重要的是，我们要验证完整的标识符集，例如，跨制片人和消费者的电影 ID 列表，以提高所选数据传输层的整体信心。对于端到端审计，目标是通过大数据平台调度程序（Big data Platform Scheduler）每小时运行一次审计，大数据平台调度程序是 Netflix 数据平台提供的一个集中集成工具，用于以高效、可靠、可重复的方式运行工作流。审计的相等性检查（即查询结果应该相同），在多次运行中两个数据集之间的对称差异应该为空，并且在 SLA 内应该最终一致性。当一组主键在真实源和目标 Data Mesh 表之间始终不匹配时，每小时都会发送一次通知。

端到端（黑盒）审计示例

人工综合事件审计是人为触发的变更事件，以模拟服务的常见 CUD 操作。它以恒定地频率生成心跳信号，目的是将其作为基线，以验证管道的健康状况，而忽略其流量模式或偶尔的静默期。

5数据消费

我们的工作室合作伙伴依靠数据来做出明智的决策，并在与制作相关的所有阶段进行协作。Studio 技术解决方案（Studio Tech Solutions，STS）团队在一些数据工具中提供了近实时的报告（我们将之称为跟踪器），以增强决策能力。

在过去的几年中，这些跟踪器中有许多都是由手动管理的 SQL 脚本和来自乐高（Lego，在 Java 服务中实现了 CRON 调度器）的 API 调用驱动的。乐高（Lego）是 STS 团队的主要工具，在巅峰时期，乐高管理着 300 多个跟踪器。

这种策略有其自身的一系列挑战：缺乏模式，并且将每个报告列视为一个字符串，这并非总是可行的，对直接 RDS 连接的依赖不稳定以及来自第三方 API 的速率限制（限流）通常会导致作业的失败。我们有一组专门为报告量身定制的“核心视图”，但这导致了即使是非常小的字段子集查询也会变得很缓慢且成本高，因为视图在这个小子集检索被执行之前做了大量的连接和聚合工作。

除了这些问题之外，当我们没有很多跟踪器同时都需要维护时，这种方法也很有效，但是随着我们创建了更多的跟踪器，达到了数百个，我们开始遇到维护、意识、知识共享和标准化方面的问题。新的团队成员很难上手，弄清楚哪个 SQL 支持哪个跟踪器是很困难的，缺乏标准使得每个 SQL 看起来都不一样，并且随着数据源的变化，必须更新跟踪器也是一场噩梦。

考虑到这一点，Studio 技术解决方案专注于构建了 Genesis，这是一个语义数据层（Semantic Data Layer），允许团队将数据源定义（Data Source Definitions ）中的数据点映射为 YAML 文件，然后根据输入定义（Input Definitions）文件中指定的选择字段、过滤器、格式化程序，使用这些数据点生成跟踪器所需的 SQL。Genesis 负责根据数据源定义中的可用内容以及用户通过执行输入定义所指定的内容来连接、聚合、格式化、过滤数据。

Genesis 数据源和输入定义示例

Genesis 是一个用 Node.js 编写的无状态 CLI，它根据参数中指定的路径从文件系统中读取它所需要的所有内容。这使得我们能够将 Genesis 集成到 Jenkins Jobs 中以提供 GitOps 和 CI 经验来维护现有的跟踪器，以及创建新的跟踪器。我们可以简单地变更数据层，触发一个空的拉取请求，查看变更，并使我们的所有跟踪器都与数据源的变更保持同步。

截至到撰写本文时，Genesis 支持了 240 多个跟踪器，并且每天都在增长，使得我们全球工作室的数千名合作伙伴都能够使用近实时的数据进行协作、注释并共享信息。