https://dagster.io/ logo
#integration-dbt
Title
# integration-dbt
e

Edson Henrique

03/15/2023, 2:42 PM
hello everyone, i'm stucked with scheduling dbt jobs using tag selection, can someone help me here? i have some jobs with this config
Copy code
{{ config(
    materialized="table",
    schema="assets",
    tags=["asset", "audiences", "metrics", "freshness_5am"]
) }}
after that, i create my dagster model with:
Copy code
dbt_morning_assets = load_assets_from_dbt_project(
    project_dir = dbt_base_cfg['project_dir'],
    profiles_dir = dbt_base_cfg['profiles_dir'],
    select = "tag:freshness_5am",
    key_prefix = ["dbt_freshness_5am"]
)

dbt_morning_assets_job = define_asset_job(
    name = "update_dbt_assets_at_5am",
    selection = AssetSelection.groups("dbt_freshness_5am")
)

dbt_morning_assets_schedule = ScheduleDefinition(
    cron_schedule="0 5 * * *",
    job=dbt_morning_assets_job,
    execution_timezone="America/Sao_Paulo"
)
but it don't find the assets, where i'm doing wrong?
🟢 1
i just used the fresshness policy method on my model i used the config
Copy code
{{ config(
    materialized="table",
    schema="dims",
    tags=["dimensions", "audiences"],
    dagster_freshness_policy={
        "maximum_lag_minutes": 60 * 5,
        "cron_schedule": "0 5 * * *",
        "cron_schedule_timezone": "America/Sao_Paulo"
    }
) }}
got the assets with
load_assets_from_dbt_project
and created a sensor
q

Qwame

03/16/2023, 1:54 AM
I think the reason it didn't work is the group name. You have to pass a function to the
_node_info_to_group_function
to extract the last element in the tag list as the group name. Then when you use
AssetSelection.groups('freshness_5am')
It will work. As you have it now, the group name is not
dbt_freshness_5am
and that's why the assets are not loading. I think it's going to be something like
dbt_freshness_5am/another-name-maybe
e

Edson Henrique

03/16/2023, 5:32 PM
@Qwame thanks alot!
3 Views