DBT 运行（或特定模型）完成后如何运行 python 代码？

Question

我希望能够运行一个临时的 python 脚本，该脚本可以访问并在 dbt 运行计算的模型上运行分析，是否有关于此的最佳实践？

Answer 1

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

这个想法是，您可以定义要在 dbt 模型运行后运行的 python 脚本：

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

iris如果模型在最后运行，则调用文件notify.py dbt run：

# 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变量中运行的当前模型。

要开始使用 fal，只需pip install fal开始编写 Python 脚本即可。