joshua
05/08/2023, 11:06 AM# pass config as arg
my_op(build_op_context(), 1, 2, MyConfig(z=3))
But how to do this when calling the op from a job?
When attempting to do the above in a job I get an error when starting up Dagster:
Must pass the output from previous node invocations or inputs to the composition function as inputs when invoking nodes during composition.
🤔Zach
05/08/2023, 2:27 PMcontext
is passed automatically under the hood - you won't need to use build_op_context
when constructing a graph or calling ops (it's generally used for testing ops that rely on an OpExecutionContext
. to pass arguments to your job to kick it off you'll generally use run config, which will allow you to pass initial configuration to each op that defines a config schema or pythonic config.sean
05/08/2023, 4:33 PMjoshua
05/09/2023, 7:04 AMjoshua
05/09/2023, 7:04 AMclass MyConfig(Config):
z: int
@op
def my_op(context, config: MyConfig):
return config.z
# pass config as arg --> works
my_op(MyConfig(z=3))
my_op(MyConfig(z=4))
# how to invoke op from job with different config values? --> doesn't work
@job
def my_job():
my_op(MyConfig(z=3))
my_op(MyConfig(z=4))
joshua
05/09/2023, 7:05 AMmy_config = RunConfig(
ops={"my_op": MyConfig(z=3)}
)
@job(config=my_config)
def my_job():
my_op()
joshua
05/09/2023, 7:24 AMsean
05/09/2023, 12:02 PMfrom dagster import Config, RunConfig, job, op
class AddNConfig(Config):
n: int
@op
def return_one(context) -> int:
return 1
@op
def add_n(context, number, config: AddNConfig) -> int:
return number + config.n
@op
def adder(context, a: int, b: int) -> int:
result = a + b
print(result)
return result
@job
def inputs_and_outputs():
value = return_one()
a = add_n(value)
b = add_n(value)
adder(a, b)
inputs_and_outputs.execute_in_process(
run_config=RunConfig({
"add_n": AddNConfig(n=1),
"add_n_2": AddNConfig(n=1)
})
)
joshua
05/09/2023, 2:21 PM