https://dagster.io/ logo
Join Slack
Powered by
Is setting resources on individual jobs supposed t...
# ask-community
k
Is setting resources on individual jobs supposed to still work when using top level resources? Even trying the example code here:
Copy code
class Server(ConfigurableResource):
  server_name: str
  def ping_server(self):
    print("pinging server " + self.server_name)

@op
def interact_with_server(context, server: Server):
  server.ping_server()

@graph
def do_stuff():
  interact_with_server()

do_stuff_staging = do_stuff.to_job(name="do_stuff_staging", resource_defs={"server": Server(server_name="staging")})
do_stuff_prod = do_stuff.to_job(name="do_stuff_prod", resource_defs={"server": Server(server_name="prod")})
When including those jobs in the Definitions I'm getting "resource with key 'server' requied by op 'interact_with_server' not provided. Please provide resource to key 'server', or change it one of the following keys [ list of all my top level resources ]
q
Did you include the resource in your Definitions?
k
no, not the top level resources, i just included them in the 
to_job
call (the 
resource_defs
parameter)
my definititions might look something like this:
Copy code
defs = Definitions(
  assets=[asset1, assetb],
  jobs=[do_stuff_staging, do_stuff_prod],
  resources=[global_resource_a, global_resource_b]
)
But I thought that should work -- without placing the resource 
server
in the definitions -- otherwise what's the point of the 
resource_defs
parameter on the 
to_job
function
c
Hi Kyle. It's a bit confusing, but top level resources provided in the 
Definitions
object are bound to assets. Jobs must already have resources bound at creation time and do not respect the resources provided top level.
q
@claire I thought resources provided in 
Definitions
were bound to jobs as well? Hence I have removed 
BindResourcestoJobs
from my project
c
Ah, yeah, you are right--the API doc I was looking at was out of date
k
I'm saying though that the top level resource binding works -- but the binding to individual jobs no longer works at all -- i'm not sure if that was intended. Like my example above just does not function, I get complaints that it cannot find the resource (and I think something similar is used as an example in the docs)
I did open a bug issue on the dagster github, but it hasn't gotten any traction
c
Hm.. what do you mean that the resource cannot be found? I'm running this example and it works locally for me:
Copy code
class Server(ConfigurableResource):
    server_name: str

    def ping_server(self):
        print("pinging server " + self.server_name)


class MyUnrelatedResource(ConfigurableResource):
    my_field: str


@op
def interact_with_server(context, server: Server):
    server.ping_server()


@graph
def do_stuff():
    interact_with_server()


do_stuff_staging = do_stuff.to_job(
    name="do_stuff_staging", resource_defs={"server": Server(server_name="staging")}
)
do_stuff_prod = do_stuff.to_job(
    name="do_stuff_prod", resource_defs={"server": Server(server_name="prod")}
)


@asset(required_resource_keys={"global_resource_a"})
def asset1():
    return 1


@asset(required_resource_keys={"global_resource_a"})
def assetb():
    return 1


# elsewhere
defs = Definitions(
    assets=[asset1, assetb],
    jobs=[do_stuff_staging, do_stuff_prod],
    resources={
        "global_resource_a": MyUnrelatedResource(my_field="alsdjaksld"),
    },
)
k
@claire sorry for slow response...Interesting, yes that small example does work. It def does not within my repo which is much bigger, but near the same structure -- (utilizing same example ops/jobs) trying to figure out how to debug this or narrow this down better
ahh, if you add an io manager to the top level resources it no longer works. So changing the defs at the end of the example to this:
Copy code
# elsewhere
defs = Definitions(
    assets=[asset1, assetb],
    jobs=[do_stuff_staging, do_stuff_prod, job_with_global_resource],
    resources={
        "global_resource_a": MyUnrelatedResource(my_field="alsdjaksld"),
        "io_manager": fs_io_manager,
    },
)
If an IO Manager is added to the jobs resource defs though, it does work though. Error message sorta had me looking in the wrong direction.
I think some docs that explained the logic of how it's choosing the top level resources vs job resources could be helpful, or maybe explicitly checking for the 'special' resources or something and giving a non misleading error? idk
dagster docs feedback 1
c
Hm.. very odd. I was able to repro the error you're reporting locally. Would you mind linking me to the github issue? I can add some comments and raise it to the team again
k
Open in Slack
PreviousNext
Powered by