在 DBT 运行（或特定模型）完成后，我如何运行 python 编码？

Question

我希望能够运行一个临时 python 脚本来访问和运行由 dbt 运行计算的模型分析，对此有什么最佳做法吗？

Answer 1

对于生产，我建议使用编排层，例如 apache airflow。

请参阅 this blog post 开始，但本质上您将拥有一个编排 DAG（注意 - 不是 dbt DAG），它执行以下操作：

dbt run <with args> -> your python code

不过，公平警告，这会给您的项目增加一点复杂性。

我想您可以使用 CI/CD 工具（例如 github 动作或 circleCI

来获得类似的效果

Answer 2

我们最近构建了一个非常适合这种情况的工具。它利用了从 Python-land 中的 dbt 引用表的便利性。它被称为fal。

我们的想法是，在您的 dbt 模型运行:

之后，您将定义您想要运行的 python 脚本

# schema.yml
models:
- name: iris
  meta:
    owner: "@matteo"
    fal:
      scripts:
        - "notify.py"

如果 iris 模型在上一个 dbt run 中是运行，则调用文件 notify.py:

# notify.py
import os
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

CHANNEL_ID = os.getenv("SLACK_BOT_CHANNEL")
SLACK_TOKEN = os.getenv("SLACK_BOT_TOKEN")

client = WebClient(token=SLACK_TOKEN)
message_text = f"""Model: {context.current_model.name}
Status: {context.current_model.status}
Owner: {context.current_model.meta['owner']}"""


try:
    response = client.chat_postMessage(
        channel=CHANNEL_ID,
        text=message_text
    )
except SlackApiError as e:
    assert e.response["error"]

每个脚本都是运行，并在 context 变量中引用了当前模型运行ning。

要开始使用 fal，只需 pip install fal 并开始编写您的 python 脚本。

在 DBT 运行（或特定模型）完成后，我如何 运行 python 编码？

How can I run python code after a DBT run (or a specific model) is completed?

python

dbt

在 DBT 运行（或特定模型）完成后，我如何运行 python 编码？