在这篇文章中，我们如何在 Apache Airflow 中调度 2000+ 个 DBT 模型？

近年来，DBT (数据构建工具) 已经确立了自己作为首选的数据转换工作流程，使用表达力强的SQL声明和Jinja模板，连接到各种处理引擎，如。它提供了强大的文档、测试支持和社区扩展包，以增强其原生功能。这无疑让ELT中的转换部分变得更加简单和愉快。

尽管DBT Core关注了模型之间的血缘关系，它并没有提供关于在生产环境中何时和何处执行的具体解决方案。换句话说，它没有自带的执行编排功能。

在这篇文章中，你将看到我们是如何使用Airflow来编排我们的DBT Core项目，创建了一个直观的管道，使数据分析师甚至产品负责人也能轻松创建和维护自己的数据模型。通过使用SQL和Git的基本知识，企业中的不同人员可以在几分钟内将他们的模型转换成具备内置警报、数据质量测试和访问控制功能的Airflow DAG，准备在分布式和可扩展的环境中运行。最重要的是：无需深入了解Airflow DAG的内部运作——除了在UI中与它互动😄

我们来把它拆分成几个关键点：

1. 单DAG vs 多DAG方法
2. 项目结构和DAG布局
3. DAG生成流程
4. 我们是如何以及为什么创建了我们的DBTOperator
5. 结论及未来方向

一种直观的方法是将整个DBT项目建模为“一个大的DAG”。这使根据DBT的数据血缘关系更容易连接任务，并在Airflow中提供整个DBT项目的清晰数据血缘图。

然而，Mono DAG 方法在我们开始这个项目时存在一些劣势，这些劣势对我们来说非常重要。

• 由于调度是在DAG层面设置的，这意味着你的整个项目将会遵循相同的调度。如果你在项目中的不同模型有不同的服务水平协议（SLA），这将是一个问题。
• 这个庞大的DAG可能很难导航。如果你的项目中有2000多个模型，在这个庞大的DAG中找到你需要的部分可能会很困难，尤其是对于那些不怎么熟悉Airflow的分析师和业务人员来说。
• 它在访问控制方面不够高效。由于我们有不同的团队负责项目中不同的部分，我们需要在Airflow中利用这种分隔：只有你的团队应该能够手动触发你的模型或决定执行完整的刷新，例如。这意味着整个项目只有一个大的DAG，只有一个访问控制层。
• 在模型失败的情况下，很难细分通知。同样，我们希望在模型失败的情况下，只通知相关的团队。

重要提示：我们在DBT原生支持多项目之前就已经开始了项目。虽然DBT Core还不完全支持这一点，但是可以通过DBT mesh来分割项目，让每个项目拥有一个DAG不再那么麻烦。

注：DAGs为“有向无环图”的简称。

为了解决上述问题，我们决定根据对我们组织有意义的分组规则将项目拆分成不同的DAG。通过这样做，我们可以为项目的不同部分设置不同的服务水平协议(SLA)，实现DAG级别的访问控制，并在回调函数中设置不同的警报/通知目标。此外，团队可以轻松地只过滤他们自己的DAG，并在Airflow中浏览自己的模型时有更好的体验。

然而，自然会有一些问题浮现，例如：我们如何将哪些模型分组到哪些有向无环图(DAG)中？我们又如何连接这些相互依赖的DAG呢？

这些问题引导我们开发了现在的解决方案。值得一提的是，能够在 Airflow 中看到完整的 DBT 线路对我们来说并不重要。我们使用 Datahub 进行数据探索，它提供了一个非常棒的线路视图。因此，我们决定将 Airflow 用于以最有效的方式管理模型的执行，而不是将它作为数据发现工具。

在思考前面提到的问题时，我们提出了模型组的概念。模型组是一组高度相关的数据转换——比如说，来自同一个数据集市，需要一起更新并且只能作为一个整体才有意义的表。此外，由同一个团队负责拥有和维护。模型组旨在实现业务中的某个目标：执行中间转换并准备一组表，创建一个数据集市，以及计算关键绩效指标等。

所以，我们决定为每个模型组对应一个DAG，因为这些模型关联紧密，并且需要一起安排运行。

下面这个简单项目结构有助于理解布局。

最简约的DBT项目结构。

我们来分步骤说说吧：

• dbt_project.yml: 这是项目根目录下常规的 dbt_project.yml 文件。没有什么特别的。
• deployment.yml: 在此文件中，您注册要部署的模型组，即将模型组转换为 DAG，并可以指定运行计划、标签、所有者等。
  该文件的内容大致如下：

    # deployment.yml  
    ---  
    model_groups:  
      - name: model_group_a # 这是文件夹的名称（例如 model_group_a）。  
        schedule: 0 0 * * * # 这是 DAG 的调度时间。  
        owner: Team_A # 这是在 Airflow 中拥有 DAG 的角色（Team_A）。  
        tags: [tag1, tag2] # 标签：Airflow DAG 的标签  
        description: 备注：这准备了进一步转换所需的表。 # DAG 描述。  

      - name: model_group_b  
        schedule: 0 2 * * *  
        owner: Team_A  
        tags: [tag1, tag2]  
        description: 备注：通过连接多个表来生成数据集市。

• model_group_a 和 model_group_b ：包含 SQL 模型的文件夹（DBT 的 Python 模型也以相同的方式工作）。比如说，model2.sql 在 model_group_a 中引用 model1.sql 作为依赖。一个 model_group 实际上只是一个 DBT 项目中的文件夹，里面包含模型。你可以根据需要在里面放入任意数量的模型，它还支持为子文件夹生成 DAG。

在 Airflow 中，这种结构如下所示。

DBT项目中的气流DAG布局。

通过这种结构，我们确保了几个重要的方面。

• 依赖的DAG通过传感器互相连接：这允许每个模型组在不同的时间表上执行，同时防止故障传播到下游。如果传感器探测失败，下游模型将被跳过。值得注意的是，我们不得不分叉原生的Airflow外部任务传感器。这是因为我们希望检查给定上游模型执行的最后一次状态，独立于其执行日期。原生传感器仅允许在特定执行日期进行探测。
• 在同一个DAG中，执行的线程基于dbt-test任务：这可以阻止数据质量问题向下传播，防止数据质量问题像滚雪球一样越积越多。
• 每个DAG（模型组）都有一个所有者：这意味着只有合适的团队成员才能对DAG执行诸如触发完全刷新运行或清除任务等手动操作。
• DAG的数量及其大小是灵活的，遵循DBT项目的布局：由于所有DAG都是根据模型组动态生成的，它们可以非常细粒度或非常大。DBT项目中的模型组所包含的模型数量决定了DAG的布局。

现在，我们将探索从团队成员在DBT仓库中创建拉取请求的那一刻开始所发生的事情。简而言之，DBT项目的部署流程如下。

有两个非常重要的要素被加入作为每个拉取请求中的CI步骤。

1. 检查组织治理要求：每个模型都需要有一个所有者和所需的标签和描述等。这非常重要，因为它充实了Datahub，使数据变得更有意义。
2. 在 staging 环境中运行更新的模型：这确保引入的更改能使更新后的模型及其所有下游依赖项顺利运行。我们的DBT CI运行的暂存区域包含了生产模型的代表性样本，降低了CI测试的成本。在CI中正确地测试DBT模型是一个独立的话题，需要专门的帖子来讨论。

在PR合并之后，我们的DAG生成过程就会启动。它通过解析DBT的manifest.json文件来生成完整的图。然后，根据deployment.yaml文件中定义的模型组规则，根据这些规则生成不同的DAG。

在这里的一个重要概念是在解析DBT清单文件时区分“内部”和“外部”模型。内部模型指的是位于相关模型组内的模型，而外部模型指的是该模型组之外的依赖模型。通过这种区分，我们可以使用我们的ExternalLatestTaskSensor（源自Airflow外部任务传感器的一个分支版本）分配适当的传感器实例。我们修改了元数据库查询，以便获取上游任务的最新状态（按执行日期排序），因此，该传感器会检查上游任务的最新dbt-test结果。

当我们解析模型组时，如何确定“内部”和“外部”模型。

因此，每个模型组通过传感器连接，使它们可以各自按计划运行。我们考虑的另外一个选择是使用 [TriggerDagRunOperator](https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/trigger_dagrun/index.html)，但这样做只能让我们在上游模型中设定运行时间。

DAG的实际生成是通过使用Jinja模板来完成的，因为我们最终只是在创建一些Python文件 😃。我们需要确定的是在特定DAG中包含哪些模型，它们的“内部”依赖关系，以及模型的“外部”依赖（如传感器）。

最后，在生成完成后，这些DAGs和DBT项目工件被推送到Airflow的工件桶，在那里另一个在Airflow中运行的进程会接收到这些工件。如果你想了解更多关于Airflow的内容，请参阅我的Airflow文章。

如果我们使用 Airflow 来运行 DBT，我们可以使用 BashOperator 来执行 dbt 命令，或者我们可以创建一个 DBTOperator 来处理这些任务。后者有很多好处，我将解释为什么你可能需要自己创建一个 DBTOperator。

我们从使用airflow-dbt项目提供的开源实现开始我们的DBTOperator之旅。那在最初的几周或几个月里运行良好，但我们意识到最好还是自己开发一个操作符（即DBTOperator），使表达更自然。

我们打算改用DBT程序化调用，它提供了更好的方法来处理运行结果，同时也符合最佳实践。在使用dbt cli的Python入口点后，代码因此变得更加整洁和易读。

最重要的是，我们想解决我们在DBT编排方案中明显的限制，特别是在手动修复错误时常常遇到的问题。以下是一些明显的限制。

原来的 DBT 原生 on_schema_change 选项 都无法解决我们的问题，因为在几乎所有的案例中，当我们添加一个字段时，我们都需要填充缺失的信息。因此，在模式发生变化时，唯一的方法是触发完整的刷新。我们最终因为预期的源模式变化导致了大量的模型失败，那时，我们只能通过删除 Snowflake 中的表来触发完整刷新。

当然，这并不是理想的情况。因此，我们在自定义的 DBTOperator 中实现的功能之一是能够在运行失败后解析 dbt-run 执行日志的能力。如果通过解析日志，发现失败是由模式变更引起的，我们就会自动重新运行这个模型 并传递 --full-refresh 参数。这个简单的功能让我们在日常维护 DBT 模型时节省了数小时。

大模型的初步处理或全面刷新

有时，在进行非常大的模型的初始处理阶段，或者因各种原因手动触发full-refresh时，我们发现这会导致我们的Snowflake DBT 仓库过载。为了避免这种情况，我们在DBTOperator中创建了一个功能，该功能可以动态更改用于执行该模型的仓库实例并设置其大小（小型、中型、大型等）。

通过这样做，我们可以在同一个DBT Warehouse上运行所有小型增量模型，同时在一个专用资源的独立仓库中执行大规模任务。这可以避免Snowflake查询积压。

此外，我们允许数据分析师触发完整的刷新操作，直接通过Airflow界面，而无需删除表。其主要功能是在适当的时候将--full-refresh标志传递给dbt run命令。

有时，我们的数据分析师需要单独触发运行 DAG 模型组中的某一个或两个模型。有时，这些运行需要对指定模型进行完全刷新。

为了适应这种情况，我们创建了作为Airflow DAG参数的选项，让分析师能够仅触发DAG中的特定模型。在特定DagRun中未被选择的所有其他模型将被跳过。这种方法通过避免运行DAG中的所有模型，仅执行一个或两个模型来防止资源浪费。

虽然使用 Airflow 的清除任务选项也是一种解决方法，但在用户需要执行完整刷新或更改执行该模型的仓库时，这种方法就显得不够用了。仅通过清除一个任务，用户无法通过参数来自定义运行，这在 Airflow 中是行不通的。这个自定义选项确保分析师可以更精确地指定他们的需求，从而提高模型执行的效率和灵活性，让分析师的工作更加高效和灵活。

在我们的Airflow-DBT结构中，我们有许多按模型组分组的DAG，并且有些模型依赖于由4-5个DAG组成的较长的依赖链。当该链中的第一个DAG中的某个模型执行（run或test）失败时，为了防止错误传播，所有下游DAG中的模型将被跳过。我们如何确保修复第一个模型后重新运行所有下游的DAG？

之前，我们是手动做的 😞。数据分析师必须跟踪并重新触发那些在模型修复后需要重新运行的下游DAG。这个过程既耗时又容易出差错。

为了解决这一问题，我们在Airflow DAG中创建了一个下游触发选项，通过DBTOperator中的自定义逻辑实现。通过在DAG执行时设置此值，如果所有模型成功，则DAG会自动识别所有下游依赖关系并触发它们的DagRun任务。这消除了修复错误后手动触发DAG的需求。

这个过程使用dbt ls命令非常简单地实现，该命令列出了在依赖图中的模型。接着，我们将它们映射到它们所在的有向无环图（DAG），并使用Airflow的[trigger_dag()](https://github.com/apache/airflow/blob/f4c4519f89fcfacf9ef3494f820e4138a2ec3d05/airflow/api/common/trigger_dag.py#L105)函数来自动触发下游任务的执行。

更重要的是，这个过程会自动以“连锁反应”的方式继续：当触发的DAG完成后，它会自动触发其下游依赖，继续这个过程，直到链中的最后一个DAG也完成。

在实现上述解决方案于DBTOperator之后，我们也创建了DAG参数，以便向用户暴露一些配置选项，让用户可以自定义手动运行。

DAG 参数可以作为 DBT 的接口使用

现在数据分析师和分析工程师可以完全自定义手动任务，这极大地改善了他们的日常工作。这样，他们可以根据需要进行自定义。

这些参数在生成DAG时会在模板中动态添加，正如前面提到的。因此，它使用该模型组中的可用模型来填充Models下拉框，例如。

另外，这种“参数注入”方法在DAG生成流程中非常灵活，让我们在将来需要时可以轻松添加更多参数。

我希望这篇帖子能带来不同的视角。尽管这种实现方式即使在两年后仍然满足我们的需求，但它还远远不够完美或理想，还有改进的空间。这种实现是在 Airflow 中进行 DBT 编排。

一个类似的开源实现是天文学家的Cosmos项目（https://github.com/astronomer/astronomer-cosmos）。这里有个有趣的功能是使用任务组将每个模型的`run`和`test`结合在一起，正如我们所做的一样 😍。以及根据项目配置简单明了地声明DbtDag的方式。

也可以将你的项目拆分成多个DAG，因为构造函数可以接受dbt select参数，所以你可以这样做。你可以通过传递不同标签来拆分项目，例如。然而，我不太清楚它们是如何处理跨DAG的交互（模型引用）的，如果有。如果你刚开始DBT编排之旅，一定不要错过这个，因为它提供了一个非常清晰的抽象。

到目前为止，我们面前的一个重要事项是实施数据合同。通过使用合同作为源系统和数据湖中表之间的纽带，我们可以自动化部署连接器（数据提取器），并且还可以自动运行DBT模型来执行基础转换，这些转换不需要特定的业务知识，如****类型转换、标准化列名以及解构复杂字段。因此，之前提到的一些模型组正在根据数据合同自动创建。

我很乐意进一步讨论DBT-Airflow的实现，非常想听听社区里大家是如何解决这个问题的。所以，如果你也有类似的实现，请告诉我你是怎么做的 😆。只有通过互相帮助的社区，我们才能打造出真正有用的好东西。