https://dagster.io/ logo
Powered by
Title
e

Eric Goddard

10/28/2022, 8:23 PM
Hi, we’re struggling to pass vars to the dbt resource in an asset job. In dagit when I launch the job with a configuration like 
resources:
  dbt:
    config:
      vars:
        refresh_periods:
        - 2022Q2
The dbt command that gets run does not have the variables. setup details are 🧵
If we have a repo with a dbt_resource configured like so: 
DBT_PROJECT_DIR = os.environ.get("DBT_PROJECT_DIR")
DBT_PROFILES_DIR = os.environ.get("DBT_PROFILES_DIR")

dbt_resource = dbt_cli_resource.configured(
    {"profiles_dir": DBT_PROFILES_DIR, "project_dir": DBT_PROJECT_DIR, "target": "prod"}
)

dbt_assets = with_resources(
    load_assets_from_dbt_project(project_dir=DBT_PROJECT_DIR, profiles_dir=DBT_PROFILES_DIR, use_build_command=True),
    {"dbt": dbt_resource},
)

@repository
def repo():
    definitions = [
        dbt_assets,
    ]

    return definitions
and a job: 
update_asset_job = define_asset_job(
    "update_asset_job",
    selection=["+some_dbt_model"],
    description="Builds dbt models",
)
shouldn’t i be able able to pass a var to the job via dagit with a configuration like this: 
resources:
  dbt:
    config:
      vars:
        refresh_periods:
        - 2022Q2
When the job runs the dbt command that gets logged is: 
Executing command: dbt --no-use-color --log-format json build --project-dir /opt/dagster/dagster_home/ --profiles-dir /opt/dagster/dagster_home/ --target prod --select dbt_project.subdir.some_early_model dbt_project.subdir.some_dbt_model
Is there something i’m missing with how the dbt cli resource is configured or how the run configuration is setup? Can vars be passed like this to a job created with 
define_asset_job
with dbt assets?
a

Adam Bloom

10/31/2022, 6:42 PM
Just responded on the dbt slack, but thought I had seen this here too! but this answers my question there - you're using assets
e

Eric Goddard

10/31/2022, 6:43 PM
yep, creating the job with 
define_asset_job
as above
a

Adam Bloom

10/31/2022, 6:45 PM
https://docs.dagster.io/_apidocs/libraries/dagster-dbt#assets - 
load_assets_from_dbt_project
has two arguments that you can use for this. 
partitions_def
and 
partition_key_to_vars_fn
. It's a bit clunky for some use cases, but it does work. I set up static partitions. You'll just need any sensors/schedules/etc to trigger with the partition key to have the vars be passed on
This was much simpler before assets came into play. cc @owen - I haven't put any thoughts into if it'd be possible to simplify this API, but not sure how intuitive the current vars methods are
e

Eric Goddard

11/01/2022, 12:23 PM
Thanks @Adam Bloom! We were able to get it working using the partitions_def and mapping function. working example for future reference: 
dbt_resource = dbt_cli_resource.configured(
    {"profiles_dir": DBT_PROFILES_DIR,
    "project_dir": DBT_PROJECT_DIR,
    "target": "prod"}
)

REFRESH_QUARTERS = [
    f"{year}Q{quarter}"
    for year in range(2020, date.today().year + 1)
    for quarter in [1, 2, 3, 4]
]

quarterly_partitions_def = StaticPartitionsDefinition(partition_keys=REFRESH_QUARTERS)

def partition_key_to_dbt_vars(partition_key):
    return {"refresh_periods": [partition_key]}

dbt_assets = with_resources(
    load_assets_from_dbt_project(
        project_dir=DBT_PROJECT_DIR,
        profiles_dir=DBT_PROFILES_DIR,
        select="+some_dbt_model +some_other_dbt_model",
        use_build_command=True,
        partitions_def=quarterly_partitions_def,
        partition_key_to_vars_fn=partition_key_to_dbt_vars,
    ),
    {"dbt": dbt_resource},
)
20 Views
#dagster-supportJoin Slack
Powered by