https://dagster.io/ logo
Title
b

Binoy Shah

10/07/2022, 12:31 PM
I am having hard time figuring out correct options to build this flow out. Between Assets/MultiAsset/Graph I feel in the weeds on how best to achieve this. Even then, the errors are super confusing. Additional Details including code and erorrs are in the 2 thread replies
:dagster-bot-resolve: 1
The code is for trying to build Graph Backed assets.
I am getting the error below
/workspace/data-engg/data-lake-workflows/venv/lib/python3.9/site-packages/dagster/_core/definitions/composition.py:91: UserWarning: While in @graph context 'perform_full_elt', received an uninvoked op 'verify_elt_data'.
  warnings.warn(warning_message.strip())
/Users/binoy/workspace/data-engg/data-lake-workflows/venv/lib/python3.9/site-packages/dagster/_core/workspace/context.py:560: UserWarning: Error loading repository location workflows:dagster._core.errors.DagsterInvalidDefinitionError: In @graph perform_full_elt, received a tuple of multiple outputs for input "program_data" (at position 0) in op invocation verify_elt_data. Must pass individual output, available from tuple: ('organization', 'program', 'program_distribution')
n

Nicolas Parot Alvarez

10/07/2022, 12:48 PM
You shouldn't need to define a graph in general, you can directly create a job from ops: https://docs.dagster.io/concepts/ops-jobs-graphs/jobs#directly-from-ops You have an issue with the connection between two of yours ops because the upstream op outputs multiple outputs, but the downstream op only takes one input. You probably need Dynamic Output + map: https://docs.dagster.io/concepts/ops-jobs-graphs/dynamic-graphs
b

Binoy Shah

10/07/2022, 2:32 PM
So over the last few weeks I was given feedback that I need to 1. Use multi_asset to manage Meltano EL process 2. Use Asset/multi asset to manage DBT transformations 3. Use Graph Backed Assets and need to be used make it a multi step flow to include by @Zach P and @claire and @sandy https://dagster.slack.com/archives/C01U954MEER/p1664466887909789?thread_ts=1664465713.664739&cid=C01U954MEER https://dagster.slack.com/archives/C01U954MEER/p1664477603691569?thread_ts=1664465713.664739&cid=C01U954MEER So as of today, I am back to using Ops instead of multi_assets to achieve this because I am still not able to orchestrate (ELT) --> (Verification) --> (DBT) by going Assets/MultiAssets route
s

sandy

10/07/2022, 3:02 PM
Hey @Binoy Shah - I believe you should be able to fix the error you're hitting by replacing
verification_results = verify_elt_data(el_results)
with
verification_results = verify_elt_data([el_results])
Happy to jump on a zoom some time to talk through what you're working on if it would be helpful
b

Binoy Shah

10/07/2022, 3:03 PM
Oh that would be tremendously helpful, please let me know your availablility
Meanwhile I am trying the suggestion you made.. Sandy
So I updated the code as recommended
@job(name="program_elt_job",
     resource_defs=meltano_resource)
def perform_full_elt():
    program = program_el()
    print(f"{program}")
    verify_elt_data([program])
but now its throwing different kind of error
/Users/binoy/workspace/data-engg/data-lake-workflows/venv/lib/python3.9/site-packages/dagster/_core/workspace/context.py:560: UserWarning: Error loading repository location workflows:dagster._core.errors.DagsterInvalidDefinitionError: In @job program_elt_job, received a list containing an invalid type at index 0 for input "program" (at position 0) in op invocation verify_elt_data. Lists can only contain the output from previous solid invocations or input mappings, received <class 'list'>
s

sandy

10/07/2022, 6:39 PM
ah - my bad - what if you try
list(program)
instead of
[program]
?
b

Binoy Shah

10/07/2022, 6:40 PM
/Users/binoy/workspace/data-engg/data-lake-workflows/venv/lib/python3.9/site-packages/dagster/_core/workspace/context.py:560: UserWarning: Error loading repository location workflows:dagster._check.ParameterCheckError: Param "resource_defs" is not one of ['Mapping']. Got <dagster._core.definitions.resource_definition.ResourceDefinition object at 0x11261ab20> which is type <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>.
s

sandy

10/07/2022, 6:41 PM
resource_defs=meltano_resource
should probably be something like
resource_defs={"meltano": meltano_resource}
b

Binoy Shah

10/07/2022, 6:46 PM
Okay that worked. Thank you
Thank you @sandy for tremendous help.