Mikael Stoltz
08/11/2022, 12:16 PM# Dev configuration
config_dbt_datamart_group_run_dev = {"models": ["DBT_DATAMART_GROUP_RUN"],
"profiles_dir": str(profile_path),
"project-dir": str(Path(curdir) / 'EDW_DQ'),
"vars" : {"prod_source_db": "EDW_DEV", "prod_source_schema":"DATAMART_GROUP"},
"target" : "dev"}
@job(resource_defs={"dbt": dbt_cli_resource.configured(config_dbt_datamart_group_run_dev)},
name='job_dbt_datamart_group_run_dev', description='This job loads the whole DATAMART_GROUP model ')
def job_dbt_datamart_group_run_dev():
dbt_run_op()
The new version: I load all assets and then in the job. Our DBT project contains several different models. I filter on the specific model in the selection part.
# Assets
dbt_odp_assets = with_resources(
load_assets_from_dbt_project(
DBT_PROJECT_DIR,
DBT_PROFILES_DIR,
source_key_prefix =["snowflake"]
),
{
"dbt": dbt_cli_resource.configured(
{"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR}
)
},
)
#Jobs
job_datamart_group_dev = define_asset_job(name="run_datamart_group_dev",
selection="QUALITY/DATAMART_GROUP_ERROR_EVENT_FACT_AGG*",
description="This job loads the whole DATAMART_GROUP model ")
Or is there a better way to do it?claire
08/11/2022, 3:53 PM@repository
def repo():
resource_defs = {
"local": {
"snowflake_io_manager": snowflake_io_manager.configured(
{
...
}
),
},
"production": {
"snowflake_io_manager": snowflake_io_manager.configured(
{
...
}
),
},
}
deployment_name = os.getenv("DAGSTER_DEPLOYMENT", "local")
return [
*with_resources(
[items, comments, stories], resource_defs=resource_defs[deployment_name]
)
]
Mikael Stoltz
08/12/2022, 6:51 AM