如何将 dbt 日志从 Airflow 发送给相关所有者？答案

【问题标题】：How do I dispatch dbt logs from Airflow to the relevant owner?如何将 dbt 日志从 Airflow 发送给相关所有者？
【发布时间】：2022-01-07 07:06:08
【问题描述】：

我正在使用KuberneterPodOperator 在Apache Airflow (v2.0.2) 上运行dbt (v0.19.0)，并在失败时向Slack 发送警报。嵌套在 dbt 中的是来自多个所有者的多个模型，它们之间存在交叉依赖关系，它们都与 run 命令一起运行。

例如：

KubernetesPodOperator(
  dbt_command="run",
  task_id="run_models",
  on_failure_callback=send_slack_alert,
)

我正在努力确保每个模型所有者都能在其相关频道中收到属于他们的警报。

为了更好地解释我的问题，假设 dbt 中有两个模型； 模型-A & 模型-B。 Model-A 归 A 团队所有，Model-B 归 B 团队所有。使用这种方法（因为有一个 dbt run 命令），如果 Model-A 出现故障，该故障将出现在 Model-A 的共享日志中，并且模型-B。我们还假设 A-Team 和 B-Team 都有自己的警报通道。但是，由于 dbt 是使用单个命令运行的，因此所有警报都会发送到一个公共通道。

现在想象有很多模型（模型-A、模型-B......模型-Z）。如何改进现有流程以确保将 Model-A 中的故障发送到 A 团队警报通道，将 Model-B 中的故障发送到 B -团队警报频道...等等。

如何将错误从 dbt（在 Airflow 中运行）分派给相关所有者以使警报可操作？

【问题讨论】：

标签： airflow dbt

【解决方案1】：

我建议您最终可能会拥有 m 个团队拥有的 n 个模型。

您最简单的更改是标记每个 dbt 模型与一个拥有团队。然后调用该模型回调该团队，例如

KubernetesPodOperator(
  dbt_command="run -m tags:team1",
  task_id="run_models",
  on_failure_callback=send_slack_alert_team1,
)

您可以考虑将参数传递给警报而不是自定义回调 (Pass other arguments to on_failure_callback)。

只要您想在所有者组中运行模型，这将起作用，但如果存在内部所有者依赖关系，则可能会出现问题。

您可以分解您的 Airflow 模型以从您的模型中组成一个动态 dag，一次运行一个模型，例如这里https://www.astronomer.io/blog/airflow-dbt-1。

然后您可以在同一个动态循环中分配 slack 运算符。

【讨论】：