19913d85-f662-4518-a9a1-7adc1793dfd4.png){kind=small}
Channels
announcements
dagster-airbyte
dagster-airflow
dagster-bigquery
dagster-cloud
dagster-cube
dagster-dask
dagster-dbt
dagster-de
dagster-ecs
dagster-feedback
dagster-kubernetes
dagster-machine-learning
dagster-noteable
dagster-releases
dagster-serverless
dagster-showcase
dagster-snowflake
dagster-spatial
dagster-support
dagster-wandb
dagstereo
data-platform-design
events
faq-read-me-before-posting
gigs-freelance
github-discussions
introductions
jobs
random
tools
豆瓣酱帮
Title
w
Will Gunadi
12/29/2021, 10:04 PMWhen an Op has one or more inputs, how does it access the context?
c
Chris Retford
12/29/2021, 10:13 PMContext is the first param.
w
Will Gunadi
12/29/2021, 10:16 PM@op
def someOp(context, first_input):✅ 1
c
Chris Retford
12/29/2021, 10:17 PMsomeOp(first_input)w
Will Gunadi
12/29/2021, 10:18 PMMake sense. I thought I tried that once and it gave me an error. I'll try again.
c
Chris Retford
12/29/2021, 10:18 PMit's currently working that way for me in 0.13.12
w
Will Gunadi
12/29/2021, 10:46 PMThis is the error I got:
In @job some_job, received invalid type <class 'str'> for input "import_ts" (at position 0) in op invocation "incremental_merge". Must pass the output from previous node invocations or inputs to the composition function as inputs when invoking nodes during composition.heres the definition:
@op
def incremental_merge(context, import_ts, table_name):incremental_merge(import_ts, table_name)@Chris Retford
c
Chris Retford
12/29/2021, 10:58 PMoh, I think I see the issue
positional inputs are for outputs from previous ops
for just vanilla params I would use either configs, or maybe *args?
or **kwargs
@op decorated functions should only have keyword arguments that match input names and, if system information is required, a first positional parameter named 'context'.you need to do something like:
@op
def get_import_ts():
# code to generate whatever import_ts is
return import_tsI would put something like table name in the config
w
Will Gunadi
12/29/2021, 11:07 PMno, because table_name is the output of the previous op
c
Chris Retford
12/29/2021, 11:07 PM@op(config_schema={"table_name":str})
def incremental_merge(context, import_ts):oh, ok
so it's the output of the previous op then
should work
w
Will Gunadi
12/29/2021, 11:08 PMdo I have to invoke it like this?
op2(op1())
op1 being the one that produce import_ts
c
Chris Retford
12/29/2021, 11:08 PMyes
I can confirm that *args and **kwargs don't work. so it looks like ops can only take outputs from other ops, and anything you put in the config
w
Will Gunadi
12/30/2021, 3:07 PM@Chris Retford I finally got it to work.
In my case I have a yield for AssetMaterialization in the Op, so I can't use return for the outputs, rather, another yield.
On top of that, I have mulitple outputs.
In the end, this is the one that works...
In the end, this is the one that works:
yield asset
yield Output('2021-12-29', 'import_ts')
yield Output('dummy', 'table_name')c
Chris Retford
12/30/2021, 3:29 PMthat is interesting. I was thinking about using AssetMaterialization. I'm just another random dev trying to figure things out 🙂
w
Will Gunadi
12/31/2021, 5:45 PM@Chris Retford I see. I have to be careful assuming that those who responded are Dagster staff 😄
It took me a while to get the AssetMaterialization working. Here's a more complete code, just to shortcut your research when you need it:
metadata = [<http://EventMetadataEntry.int|EventMetadataEntry.int>(
value=nrc, label=f"New Rows Count"
)]
asset = AssetMaterialization(
description=f"BB dummy", metadata_entries=metadata,asset_key=f'bb_repository.dummy_load',
)
yield asset
yield Output('2021-12-29', 'import_ts')
yield Output('dummy', 'table_name')