AI 不应该浪费时间重新发明 ETL

人工智能最近的进展非常令人兴奋。人们正在以各种新颖的方式使用它，从改善客户支持体验、 编写和运行代码，到制作新音乐，甚至加速医学成像技术。

但在此过程中，出现了一个令人担忧的趋势：人工智能社区似乎正在重塑数据移动（又名 ETL）。无论他们将其称为连接器、提取器、集成、文档加载器还是其他名称，人们都在编写相同的代码来从相同的 API、文档格式和数据库中提取数据，然后将它们加载到矢量数据库或法学硕士索引中。

问题在于，从头开始构建和维护强大的提取和装载管道是一项巨大的承诺。该领域有太多的现有技术，对于人工智能领域的几乎所有工程师或公司来说，重建它都是浪费大量时间。在大约每小时都会出现突发新闻的领域，主要关注点应该是让你的核心产品对用户来说令人难以置信，而不是进行支线任务。对于几乎所有人来说，核心产品不是数据移动；而是数据移动。这是你正在酿造的人工智能魔法酱。

关于构建强大的提取、转换和加载 (ETL) 管道所涉及的挑战，已经有很多文章 ( 1 , 2 )，但让我们将其放在 AI 的背景下。

为什么人工智能需要数据移动？

受过公共数据培训的法学硕士很棒，但你知道什么更好吗？人工智能可以回答我们、我们的公司和我们的用户的特定问题。如果 ChatGPT 能够了解我们整个公司的 wiki，阅读我们所有的电子邮件、Slack 消息、会议记录和文字记录，插入我们公司的分析环境，并在回答我们的问题时使用所有这些来源，我们都会很高兴。或者，当将人工智能集成到我们自己的产品中时（例如使用 Notion AI ） ，我们希望应用程序的人工智能模型在帮助用户时了解我们拥有的有关用户的所有信息。

数据移动是这一切的先决条件。

无论您是微调模型还是使用检索增强生成 (RAG)，您都需要从数据所在的位置提取数据，将其转换为模型可消化的格式，然后将其加载到 AI 应用程序可以访问的数据存储中来服务您的用例。

上图说明了使用 RAG 时的情况，但您可以想象，即使您不使用 RAG，基本步骤也不太可能改变：您需要提取、转换和加载数据（也称为 ETL），以便构建了解特定于您和您的用例的非公开信息的人工智能模型。

为什么数据移动困难？

构建一个基本的功能 MVP 以从 API 或数据库提取数据通常（尽管并非总是）可以在快速（不到 1 周）的情况下完成。真正困难的部分是使其做好生产准备并保持这种状态。让我们看看构建提取管道时想到的一些标准挑战。

增量提取和状态管理

如果您有任何有意义的数据量，则需要实现增量提取，以便您的管道仅提取以前未见过的数据。为此，您需要有一个持久层来跟踪每个连接提取的数据。

瞬态错误处理、退避、故障恢复、气隙

上游数据源始终存在，有时没有任何明确的原因。您的管道需要对此具有弹性，并使用正确的退避策略重试。如果故障不是那么短暂（但仍然不是您的错），那么您的管道需要具有足够的弹性，以记住它停止的位置，并在上游修复后从同一位置恢复。有时，来自上游的问题非常严重（例如 API 从记录中删除一些关键字段），您希望完全暂停整个管道，直到您检查正在发生的情况并手动决定要做什么。

识别并主动修复配置错误

如果您正在构建数据提取管道来提取客户的数据，则需要实施一些防御性检查，以确保客户为您提供的代表他们提取数据的所有配置都是正确的，如果不正确，请尽快执行给他们可操作的错误消息。大多数 API 并不容易做到这一点，因为它们不会发布全面的错误表，即使发布了，它们也很少为您提供可用于检查分配给 API 令牌等权限的端点，因此您必须找到平衡全面错误表的方法检查并为用户提供快速反馈。

身份验证和秘密管理

API 的简单性范围很广，从简单的不记名令牌身份验证到会话令牌或一次性令牌 OAuth 的“创意”实现。您需要实现逻辑来执行身份验证以及管理可能每小时刷新一次的机密，从而可能协调多个并发工作人员之间的机密刷新。

优化提取和加载速度、并发性和速率限制

说到并发工作人员，您可能希望实现并发来实现提取的高吞吐量。虽然这对于小型数据集可能并不重要，但对于较大的数据集绝对至关重要。即使 API 发布了官方速率限制，您也需要凭经验找出最佳并行参数，以最大限度地发挥 API 提供给您的速率限制，而不会将 IP 列入黑名单或永远受到速率限制。

适应上游 API 变化

API 一直在变化并呈现新的未记录的行为或怪癖。许多供应商每季度发布新的 API 版本。您需要密切关注所有这些更新可能如何影响您的工作，并投入工程时间来使其保持最新状态。新的端点一直在出现，并且有些端点会改变它们的行为（而且您并不总是能得到提醒）。

调度、监控、日志记录和可观察性

除了从特定 API 提取数据的代码之外，您可能还需要构建一些供所有数据提取器利用的横向功能。您将需要一些调度以及日志记录和监视，以便在调度不起作用或其他事情出现问题而您需要进行调查时进行记录。您可能还需要一些可观察性，例如昨天、今天、上周等提取了多少记录……以及它们来自哪些 API 端点或数据库表。

数据阻塞或散列

根据您从何处提取数据，您可能需要一些隐私功能来阻止或散列列，然后再将它们发送到下游。

需要明确的是，如果您只想一次性移动一些文件，则上述内容不适用。

但当您构建需要数据移动的产品时，它确实适用。迟早，您将需要处理其中的大部分问题。虽然其中没有一项是不可逾越的火箭科学，但将它们结合在一起可以快速增加一个或多个全职工作，而且您获取的数据源越多，情况就更是如此。

这正是维护数据提取和管道的困难：其大部分成本来自保持这些管道功能和稳健所需的持续增量投资。对于大多数人工智能工程师来说，这并不是给用户带来最大价值的工作。他们的时间最好花在其他地方。

那么，人工智能工程师需要做什么才能在这里移动一些数据呢？

如果您发现自己需要数据提取和加载管道，请尝试现有的解决方案，而不是自动构建自己的解决方案。他们很可能可以解决很多问题（即使不能解决您的全部问题）。如果没有，请自行构建作为最后的手段。

即使现有平台不支持您所需的一切，您仍然应该能够通过可移植且可扩展的框架获得大部分支持。这样，您无需从头开始构建所有内容，而是可以利用平台中的现成功能完成 90% 的工作，并且只需构建和维护最后 10%。最常见的例子是长尾集成：如果平台没有提供与您需要的 API 的集成，那么一个好的平台将使编写一些代码甚至无代码解决方案来构建该集成变得容易，并且仍然可以获得该平台提供的所有有用功能。即使您希望灵活地将连接器作为 python 包导入并根据您的代码触发它，您也可以使用众多开源 EL 工具之一，例如 Airbyte 或 Singer 连接器。

需要明确的是，数据移动尚未完全解决。在某些情况下，现有解决方案确实无法满足要求，您需要构建新颖的解决方案。但这并不是人工智能工程师的大多数。大多数人不需要一遍又一遍地重建与 Jira、Confluence、Slack、Notion、Gmail、Salesforce 等的相同集成。让我们使用已经经过实战测试并开放给任何人使用的解决方案，这样我们就可以继续增加用户真正关心的价值。