How do I pass a configuration value to multiple ops in a job? (job-level config)

[sryza]

[sryza]

This is a question that has come up in a few different places, so wanted to document the pattern we use currently:

@resource(config_schema=str)
def my_str_resource(init_context):
    return init_context.resource_config

@op(required_resource_keys={"my_str"})
def op1(context):
    my_str = context.resources.my_str
    # do stuff

@op(required_resource_keys={"my_str"})
def op2(context):
    my_str = context.resources.my_str
    # do stuff

@job(resource_defs={"my_str": my_str_resource})
def my_job():
    op1()
    op2()

my_job.execute_in_process(run_config={"resources": {"my_str": {"config": "some_value"}}})

[yuhan]

You can also use the built-in util method make_values_resource to pass the values via a resource and reference that resource from any op that needs it.

See example here: https://docs.dagster.io/concepts/configuration/config-schema#passing-configuration-to-multiple-ops-in-a-job

[zyd14]

You can also do this with the ConfigurableResource pythonic resource system by making a resource which only handles configuration and passing that resource to each of the different ops / assets that need to access a shared or "global" config:

from dagster import ConfigurableResource, op, graph, RunConfig, OpExecutionContext

class MyConfig(ConfigurableResource):
    val: str = "default"
    
@op
def foo(context: OpExecutionContext, config: MyConfig):
    context.log.info(config.val)

@op
def bar(context: OpExecutionContext, config: MyConfig):
    context.log.info(config.val)

@graph
def my_graph():
    foo()
    bar()

my_job = my_graph.to_job(
    config=RunConfig(
        resources=MyConfig(val="something")
    ),
)

[yuhan]

Thanks @zyd14! This is more up-to-date than the other answers, so I'm going to mark this as the answer!

[craustin]

In v1.5.8, this example code gives me:

ParameterCheckError: Param "resources" is not a dict.

When I change it to resources={"config": MyConfig(...)} instead, I get error:

DagsterInvalidConfigError: Invalid default_value for Field. Error 1: Received unexpected config entry "config" at the root. Expected: "{ io_manager?: { config?: Any } }".

[zyd14]

@craustin This seems to work with 1.5.8:

from dagster import ConfigurableResource, op, graph, RunConfig, OpExecutionContext


class MyConfig(ConfigurableResource):
    val: str = "default"


@op()
def foo(context: OpExecutionContext, config_resource: MyConfig):
    context.log.info(config_resource.val)


@op()
def bar(context: OpExecutionContext, config_resource: MyConfig):
    context.log.info(config_resource.val)


@graph
def my_graph():
    foo()
    bar()


my_job = my_graph.to_job(
    resource_defs={"config_resource": MyConfig(val="blah")},
)

[craustin]

Thanks- that change from config= to resource_defs= fixes it.

Also, the rename of the argument from config to config_resource fixes an issue where the launchpad tries to set the resource config as config on the ops.

[zyd14]

@yuhan - it seems like this might be a better answer moving forward (1.5.8+):

from dagster import ConfigurableResource, op, graph, RunConfig, OpExecutionContext


class MyConfig(ConfigurableResource):
    val: str = "default"


@op()
def foo(context: OpExecutionContext, config_resource: MyConfig):
    context.log.info(config_resource.val)


@op()
def bar(context: OpExecutionContext, config_resource: MyConfig):
    context.log.info(config_resource.val)


@graph
def my_graph():
    foo()
    bar()


my_job = my_graph.to_job(
    resource_defs={"config_resource": MyConfig(val="blah")},
)

[watsy0007]

https://docs.dagster.io/concepts/ops-jobs-graphs/op-jobs#config-mapping
i follow the document. and use configmapping to multiple ops

Instead of needing to configure every op and resource individually when launching the op job, you can supply a smaller number of values to the outer config. The ConfigMapping will then translate it into config for all the job's ops and resources.

[skeller88]

@zyd14 which of the ConfigurableResource / ConfigMapping approaches does the Dagster team recommend?

[skeller88]

Hi, just following up on this question. I'm also looking for a solution to set an op config using values from an upstream asset?

[zyd14]

@skeller88

which of the ConfigurableResource / ConfigMapping approaches does the Dagster team recommend?

I'm not part of the Dagster team, but from my perspective it seems that ConfigMappings have become a bit obsolete in that they don't play very nice with pythonic configuration, which should generally be preferred over config_schemas. With that in mind I personally prefer ConfigurableResources for this. ConfigMappings can also be bit of a pain to maintain, particularly if your config interfaces need to change.

Hi, just following up on this question. I'm also looking for a solution to set an op config using values from an upstream asset?

Configuration is set before a job is run, and is not able to be modified during an in-progress run by any of the ops / assets that are part of a given job. In other words, run config is static at runtime and can't be modified during the run. If you have a downstream asset or op that depends on a value calculated by an upstream asset or op, then that should be passed as an input to the downstream asset / op. Also, mixing ops and assets can lead to unexpected results if you're not using load_asset_value correctly, so do be careful if you're trying to use assets within a graph definition with other ops.