BC A
06/06/2023, 11:03 AMdef build_output_asset (asset_name: str, key_prefix: list, ins: dict) -> asset:
@asset(
name = asset_name,
key_prefix = key_prefix,
required_resource_keys = {"custom_io_manager"},
io_manager_key = "custom_io_manager",
group_name = "outputs",
ins = {"upstream": AssetIn(**ins)}
)
def asset_def(context, upstream) -> None:
print("WRITING IN GCS")
return asset_def
Resources definition:
resources_definitions = {
"dbt": dbt_cli_resource.configured({
"project_dir": DBT_PROJECT_PATH,
"profiles_dir": DBT_PROFILES
}),
"custom_io_manager": bigquery_pandas_io_manager.configured({
"project": GCP_PROJECT
}),
"io_manager": FilesystemIOManager(),
}
Error when I run materialization :
dagster._core.errors.DagsterExecutionLoadInputError: Error occurred while loading input "upstream" of step "dbt_mf__weather_gcs"::
FileNotFoundError: [Errno 2] No such file or directory: '/path/to/a/directory/storage/another_path/path/path'
I confirm that when I replace the custom_io_manager as the default IO manager my pipeline works but it messes my code a little bit .
Troubleshot a bit further,
basically the loaded input is using the key "io_manager" and the output is using the "custom_io_manager", isn't there a way to force using the same "custom_io_manager" in both operations ?
Thanks in advancesandy
06/06/2023, 3:22 PMBC A
06/06/2023, 4:49 PMsandy
06/06/2023, 8:28 PM