[译]可维护的 ETL：使管道更容易支持和扩展的技巧

任何数据科学项目的核心是...噔噔噔...数据！以可靠和可重复的方式准备数据是该过程的基本部分。如果你正在培训一个模型，计算分析，或者只是将来自多个源的数据组合到另一个系统中，那么你将需要构建一个数据处理或 ETL管道。

我们 Stitch Fix 这里从事的是全栈数据科学。这意味着我们以数据科学家的身份负责项目的构思、生产以至维护的整个过程。我们的受好奇心驱使喜欢快速行动，即使我们的工作常常是相互联系的。我们所处理的问题具有挑战性，因此解决方案可能很复杂，但我们不想在不需要的地方引入复杂性。因为我们必须支持我们在生产中的工作，所以我们的小团队分担随叫随到的责任，并帮助支持彼此的管道。这让我们可以做一些重要的事情，比如度假。今年夏天，我和妻子要去意大利度蜜月，这是我们多年前的打算。当我在那里的时候，我最不想考虑的是我的队友们是否很难使用或理解我写的管道。

让我们也承认数据科学是一个动态的领域，所以同事们会转向公司之外的新计划、团队或机会。虽然一个数据管道可能由一个数据科学家构建，但在其生命周期中，它通常由多个数据科学家支持和修改。像许多数据科学团体一样，我们来自不同的教育背景，不幸的是，我们并非都是“独角兽” —— 软件工程、统计和机器学习方面的专家。

虽然我们的算法小组确实有一个庞大的、令人惊叹的数据平台工程师团队，它们不会也不想写 ETL 来支持数据科学家的工作。相反，他们将精力集中在构建易于使用、健壮可靠的 工具 上，这些工具使数据科学家能够快速构建 ETL、培训和评分模型，以及创建性能良好的 API，而无需担心基础设施。

多年来，我发现了一些有助于使我的 ETL 更易于理解，维护和扩展的关键做法。本文会带大家看看以下做法有什么好处：

1. 建立一系列简单的任务。
2. 使用工作流程管理工具。
3. 尽可能利用 SQL。
4. 实施数据质量检查。

讨论细节之前，我要承认一点：没有一套构建 ETL 管道的最佳实践。这篇文章的重点是数据科学环境，其中有两件事情是正确的：支持人员的组成状况的演变是不断发展和多样化的，开发和探索优先于铁定的可靠性和性能。

建立一系列简单的任务

使 ETL 更容易理解和维护的第一步是遵循基本的软件工程实践，将大型和复杂的计算分解为具有特定目的的离散、易于消化的任务。类似地，我们应该将一个大型ETL管道划分为较小的任务。这有很多好处：

1. 更容易理解每个任务：只有几行代码的任务更容易审查，因此更容易吸收处理过程中的任何细微差别。

2. 更容易理解整个处理链：当任务具有明确定义的目的并且命名正确时，审阅者可以专注于更高级别的构建块以及它们如何组合在一起而忽略每个块的细节。

3. 更容易验证：如果我们需要对任务进行更改，我们只需要验证该任务的输出，并确保我们遵守与此任务的用户/调用者之间的任何“约定”（例如，结果表的列名称和数据类型与预修订格式相匹配）。

4. 提升模块化程度：如果任务具有一定的灵活性，则可以在其他环境中重用它们。这减少了所需的总代码量，从而减少了需要验证和维护的代码量。

5. 洞察中间结果：如果我们存储每个操作的结果，当出现错误时，我们将更容易调试管道。我们可以查看每个阶段，更容易找到错误的位置。

6. 提高管道的可靠性：我们将很快讨论工作流工具，但是将管道分解为任务的话，发生临时故障时就可以更轻松地自动重新运行任务。

我们从一个简单的示例，就可以看到将管道拆分为较小任务的好处。在 Stitch Fix，我们可能想知道发送给客户的物品当中，“高价”物品所占的比例。首先，假设我们已经定义了一个存储阈值的表。请记住，阈值将根据客户群（例如孩子与女性）和物品种类（例如袜子与裤子）而有所不同。

由于此计算相当简单，我们可以对整个管道使用单个查询：

WITH added_threshold as (
  SELECT
    items.item_price,
    thresh.high_price_threshold
  FROM shipped_items as items
  LEFT JOIN thresholds as thresh
    ON items.client_segment = thresh.client_segment
      AND items.item_category = thresh.item_category
), flagged_hp_items as (
  SELECT
    CASE
      WHEN item_price >= high_price_threshold THEN 1
      ELSE 0
    END as high_price_flag
  FROM added_threshold
) SELECT
    SUM(high_price_flag) as total_high_price_items,
    AVG(high_price_flag) as proportion_high_priced
  FROM flagged_hp_items
复制代码

这第一次尝试实际上相当不错。它已经通过使用公共表表达式（CTE）或 WITH 块进行了模块化。每个块都用于特定目的，它们简短且易于吸收，并且别名（例如 added_threshold ）提供足够的上下文，以便审阅者可以记住块中所完成的操作。

另一个积极方面是阈值存储在单独的表中。我们可以使用非常大的 CASE 语句对查询中的每个阈值进行硬编码，但这对于审阅者来说很快就会变得难以理解。它也很难维护，因为我们只要想更新阈值，就必须更改此查询以及使用相同逻辑的任何其他查询。

虽然这个查询是一个良好的开端，但我们可以改进实现的方式。最大的不足是我们无法轻松访问任何中间结果：整个计算只需一次操作即可完成。你可能想知道，为什么我要查看中间结果？中间结果允许你进行即时调试，获得实施数据质量检查的机会，并且可以证明在其他查询中可重用。

例如，假设企业添加了一个新的物品类别 —— 例如，帽子。我们开始销售帽子，但我们忘记更新阈值表。在这种情况下，我们的聚合指标就会漏掉高价的帽子。由于我们使用了 LEFT JOIN，因为连接不会删除行，但是 high_price_threshold 的值将为 NULL。到了下一个阶段，所有和帽子有关的行，其 high_price_flag 的值都会是零，而这个数值会带到我们最终进行计算的 total_high_price_items 和 proportion_high_priced 。

如果我们将这个大的单个查询分解为多个查询并分别编写每个阶段的结果，我们就可以使这个管道更易于维护。如果我们将初始阶段的输出存储到单独的表中，我们可以轻松检查我们是否没有丢失任何阈值。我们需要做的就是查询此表并选择 high_price_threshold 值为 NULL 的行。如果什么都没有返回，就代表我们遗漏了一个或多个阈值。我们将在帖子后面介绍这种类型的数据运行时验证。

这种模块化的实现也更容易修改。假设我们不是要考虑所有曾寄出的物品，而是决定只想计算过去 3 个月发送的高价物品。要是用原来的查询方式，我们就会对第一阶段进行更改，然后查看最终得出的总数，期望得到正确的数值。通过单独保存第一阶段，我们可以添加一个具有发货日期的新列。然后，我们可以修改查询并验证结果表中的发货日期是否都在我们预期的日期范围内。我们还可以将我们的新版本保存到另一个位置并执行“数据差异”以确保我们正在删除正确的行。

最后一个示例将此查询拆分为单独的阶段带来了最大的好处之一：我们可以重用我们的查询和数据来支持不同的用例。假设一个团队想要过去 3 个月的高价项目指标，但另一个团队仅在最后一周需要它。我们可以修改第一阶段的查询以支持这些并将每个版本的输出写入单独的表。如果我们为后期查询动态指定源表，相同的查询将支持两种用例。此模式也可以扩展到其他用例：具有不同阈值的团队，按客户端细分和项目类别细分的最终指标与汇总。

我们通过创建分阶段管道进行了一些权衡。其中最大的一个是运行时性能，尤其是当我们处理大型数据集时。从磁盘读取和写入数据会造成很大的开销，并且在每个处理阶段，我们读取前一阶段的输出并写出结果。和旧的 MapReduce 范例相比，Spark 的一大优势是临时结果可以缓存在工作节点（执行程序）的内存中。Spark 的 Catalyst 引擎还优化了 SQL 查询和 DataFrame 转换的执行计划，但它优化时无法跨越读/写边界。这些分阶段管道的第二个主要限制是它们使创建自动化集成测试变得更加困难，这涉及测试多个计算阶段的结果。

有了 Spark，就可以解决这些不足之处。如果我必须执行几个小的转换并且我想要保存中间步骤的选项，我就会创建一个管理程序脚本，这个脚本只有在设置了命令行标志时才执行转换，以及输出中间表。当我正在开发和调试更改时，我可以使用该标志来生成验证新计算是否正确所需的数据。一旦我对我的更改有信心，我可以关闭标记以跳过编写中间数据。

使用工作流程管理工具

使用可靠的工作流管理和调度引擎，可以实现巨大的生产力提升。一些常见的例子包括Airflow、 Oozie 、 Luigi 和 Pinball 。这项建议需要时间和专业知识来建立；这不是个别数据科学家可能负责管理的事情。在 Stitch Fix，我们开发了自己的专有工具，由我们的平台团队维护，数据科学家用它就可以创建、运行和监控我们自己的工作流程。

工作流工具可以轻松定义计算的有向非循环图（DAG），其中每个子任务都依赖于任何父任务的成功完成。这些工具通常能让使用者得以指定运行工作流的计划，在工作流启动前等待外部数据依赖，重试失败的任务，在失败时恢复执行，在发生故障时创建警报，以及运行不相互依赖的任务在平行下。这些功能相结合，使用户能够构建可靠，高性能且易于维护的复杂处理链。

尽可能利用SQL

这可能是我提出的最具争议性的建议。即使在 Stitch Fix 中，也有许多数据科学家反对 SQL，而是提倡使用通用编程语言。不久之前我还是这个阵营的一员。在实践方面，SQL 很难测试 — 特别是通过自动化测试。如果你来自软件工程背景，那么测试的挑战可能会让你觉得有足够的理由来避免使用 SQL 。我在过去也陷入过关于 SQL 的情感陷阱：“SQL 技术性较差，专业性较差； 真正 的数据科学家应该编码。”

SQL 的主要优点是所有数据专业人员都能理解：数据科学家、数据工程师、分析工程师、数据分析师、数据库管理员和许多业务分析师。这是一个庞大的用户群，可以帮助构建，审查，调试和维护 SQL 数据管道。虽然 Stitch Fix 没有很多这些数据角色 ，但 SQL 是我们这些不同数据科学家的共同语言。因此，利用 SQL 可以减少对团队中专业角色的需求，这些团队具有强大的 CS 背景，为整个团队创建管道，无法公平地分担支持职责。

通过将转换操作编写为 SQL 查询，我们还可以实现可伸缩性和某种级别的可移植性。使用适当的 SQL 引擎，可以用相同的查询语句来处理一百行数据，然后针对太字节数量级的数据运行。如果我们使用内存处理软件包（如 Pandas）编写相同的转换操作，那么随着业务或项目的扩展，我们将面临超出处理能力的风险。所有东西运行起来都不会有问题，但一到了数据集过大、内存无法容纳时，就会出错。如果这项工作正在进行中，这可能导致急于重写事情以使其恢复运行。

不同 SQL 语言变体有很多共通之处，我们从一个 SQL 引擎到另一个 SQL 引擎具有一定程度的可移植性。在 Stitch Fix 中，我们使用 Presto 进行 adhoc 查询，使用 Spark 进行生产管道。当我构建一个新的 ETL 时，我通常使用 Presto 来理解数据的结构，并构建部分转换。一旦这些部件到位，我几乎总是用 Spark运行相同的查询语句，不作任何修改。如果我要切换到 Spark 的 DataFrame API，我需要完全重写我的查询。反过来同样可以体现这种可移植性的好处。如果生产作业存在问题，我可以重新运行相同的查询并添加过滤器和限制以将数据的子集拉回以进行目视检查。

当然，不是所有操作都能用 SQL 完成。你将不会使用它来训练机器学习模型，而且还有许多其他情况下，SQL 实现即使可行，也会过于复杂。对于这些任务，你绝对应该使用通用编程语言。如果你遵循关键的建议，把你的工作分成小块，那么这些复杂的任务将在范围内受到限制，并且更容易理解。在可能的情况下，我尝试在一系列简单准备阶段的末尾隔离复杂的逻辑，例如：连接不同的数据源、过滤和创建标志列。这使得验证进入最后一个复杂阶段的数据变得容易，甚至可以简化一些逻辑。一般来说，我在本篇文章的其余部分已经不再强调自动化测试，但处理有复杂逻辑的任务时，着力实现测试覆盖就很有意义了。

实施数据质量检查

要验证复杂的逻辑时，自动单元测试非常有用，但对于作为分阶段管道的一部分的相对简单的转换，我们通常可以手动验证每个阶段。就 ETL 管道而言，自动化测试提供了混合的好处，因为它们不会覆盖最大的错误来源之一：我们的管道上游的故障导致我们的初始依赖关系中出现旧的或不正确的数据。

一个常见的错误来源是在启动管道之前未能确保我们的源数据已更新。例如，假设我们依赖于每天更新一次的数据源，并且我们的管道在数据源更新之前就开始运行。这意味着我们要么用的是（前一天计算的） 旧数据，要么使用旧数据和当前数据的混合数据。这种类型的错误可能难以识别和解决，因为上游数据源可能在我们获取旧版本的数据后不久就完成更新。

上游故障还可能导致源数据中出现错误数据：字段计算错误，模式更改和/或缺失值频率更高。在动态且互联的环境中，利用另一个团队创建的数据源进行实验的做法并不少见，而这些源也常常会出现意外更改；我们在 Stitch Fix 运作时所处的环境很大程度上就是如此单元测试通常不会标记这些故障，但可以通过运行时验证（有时称为数据质量检查）来发现它们。我们可以编写单独的 ETL 任务，如果我们的数据不符合我们期望的标准，它们将自动执行检查并引发错误。上面提到了一个简单的例子，其中缺少高价的帽子门槛。我们可以查询组合出货物品和高价阈值表，并查找缺少阈值的行。如果我们找到任何行，我们可以提醒维护者。这个想法可以推广到更复杂的检查：计算零分数、平均值、标准差、最大值或最小值。

在特定列的缺失值高于预期的情况下，我们首先需要定义预期的内容，这可以通过查看上个月每天丢失的比例来完成。然后我们可以定义触发警报的阈值。这个想法可以推广到其他数据质量检查（例如，平均值落在一个范围内），我们可以调整这些阈值，使我们对警报的敏感度进行增减。

正在进行的工作

在这篇文章中，我们已经完成了几个实际步骤，可以使你的ETL更易于维护，扩展和生产支持。这些好处可以扩展到你的队友以及你未来的自我。虽然我们可以为构建良好的流水线而感到自豪，但编写ETL并不是我们进入数据科学的原因。相反，这些是工作的基本部分，使我们能够实现更大的目标：构建新模型，为业务提供新见解，或通过我们的API提供新功能。建造不良的管道不仅需要时间远离团队，还会给创新带来障碍。

我在上一份工作中尝到的苦果，让我明白到管道如果 难以使用 ，就会让项目难以维护和扩展。我当时在某个创新实验室工作，该实验室率先使用大数据工具来解决组织中的各种问题。我的第一个项目是建立一条管道来识别信用卡号被盗的商家。我构建了一个使用 Spark 的解决方案，由此产生的系统在识别新的欺诈活动方面非常成功。然而，一旦我把它传递到信用卡部门支持和扩展，问题就开始了。我在编写管道时打破了我列出的所有最佳实践：它包含一个执行许多复杂任务的作业，它是用 Spark 编写的，当时对公司来说是新的，它依赖于 cron 进行调度并且没有'发生故障时发送警报，它没有任何数据质量检查，以确保源数据是最新的和正确的。由于这些缺陷，管道没有运行的时间延长。尽管有一个广泛的路线图来增加改进，但由于代码很难理解和扩展，因此很少能够实现这些改进。最终，整个管道以一种更容易维护的方式重写

就像你的 ETL 正在进行的数据科学项目一样，你的管道永远不会真正完整，应该被视为永远不断变化。通过每次更改，每次更改都是实现小幅改进的契机：提高可读性，删除未使用的数据源和逻辑，或简化或分解复杂的任务。这些建议并不是什么重大突破，但如果要始终如一地践行，就需要自律。就像狮子驯服一样，当管道很小时，它们相对容易控制。然而，它们长得越大，就越难管控，也越容易表现出突发且意外的错乱行为。到了那种地步，你只得重新开始、采取更好的做法，不然就可能会冒着失败的风险 [5][#f5]。

注释

提取、转换和加载的缩写。

最简单的方法是使用简单的字符串替换或字符串插值，但是你可以通过模板处理库（如 jinja2）实现更大的灵活性。

对于 Python，像标准库中的Click、Fire，甚至argparse 这样的库可以轻松定义这些命令行标志。

操作日期和从 JSON 中提取字段等操作需要修改查询，但这些更改很微小。

在撰写博客时，没有狮子或数据科学家受到伤害。

如果发现译文存在错误或其他需要改进的地方，欢迎到 掘金翻译计划 对译文进行修改并 PR，也可获得相应奖励积分。文章开头的 本文永久链接 即为本文在 GitHub 上的 MarkDown 链接。

掘金翻译计划 是一个翻译优质互联网技术文章的社区，文章来源为掘金 上的英文分享文章。内容覆盖 Android 、 iOS 、 前端 、 后端 、 区块链 、 产品 、 设计 、 人工智能 等领域，想要查看更多优质译文请持续关注 掘金翻译计划 、官方微博、 知乎专栏 。