Hey community, I’m trying to figure out a way to control the flow with the caches. Suppose I have a flow with a dependency graph like this: A_heavy_mapped_task --> B_heavy_reduce_task --> C_some_other_task Now, suppose I have the results for tasks A & B cached in a

Result

and I just want to run task C. In Luigi, C would ask for B’s output, and only fetch that. However, in prefect, the flow runner will fetch ALL the results of the mapped tasks A, but since we already have B’s result, that’s totally redundant and wastes a lot of time and data exchange. Any idea on how to tackle this issue? i.e. fetch only task B results somehow so that C can run?

Hi @Avi A! I think you've identified a special case we need to address in our mapping/caching interactions. Your desired behavior definitely sounds like something we'd like to implement. Could I persuade you to open a Github issue so we can track and prioritize this?

Running subgraphs is also on our radar, so watch this space on that front. In the meantime, decomposing this into several flows could do the trick as well. If you're using an orchestration layer (Server, Cloud), you could use that to check the state of your "upstream" flows.

yes I’m using a server. I’m not sure how I can do this, can you point me to the relevant part in the docs?

I'm not sure we have docs on that, but will put that in our queue. The interactive API in the UI is also a solid source for getting more acquainted with our API. If I'm understanding your use case correctly, in this case you'd separate task A into Flow A and tasks B and C into Flow BC. Then if you wanted to rerun just tasks B and C, you'd rerun Flow BC. If you wanted to manage BC in relation to A, you could put logic in Flow BC that queried server for the state of A or whatever criteria you'd like.

cool! how do I query the server for the state of A? I’m guessing I need to use the interactive API? can I somehow fetch the whole

state

of Flow A?

Here's an example of the sort of query you'd want to run:

query {
  flow_run (where: {flow_id: {_eq: "flow-a-id-here"}}){ 
    state
  }
  }

You can play around with this via the interactive API, but if there's something you'd like to query programmatically, you can do so using Core's

client.graphql

method.

thanks. I’m not familiar with graphql, but I see that there’s a way to use variables. how do I use the

variables

when running a graphql? I mean, what’s the mark for a variable placeholder inside a query?

Is something like this what you're thinking of? https://github.com/PrefectHQ/prefect/blob/master/src/prefect/client/client.py#L853

well I was referring the the

variables

argument in the

client.graphql

method

prefect.client.client.Client.graphql(query, raise_on_error=True, headers=None, variables=None, token=None)

Gotcha.

get_task_run_info

is essentially a canned query, so you'll want to specify your own fields. In this case, sounds like you'll want something like this:

query {
  flow_run (where: {flow_id: {_eq: "your-id-here"}}){ 
    task_runs {
      states {
        result
        serialized_state
      }
    }
  }
  }

yup, I’m just realizing this by looking at the schemas themselves

As for using variables, GraphQL allows for the use of input objects. If you'd like to use them, you can specify your query like the one linked, then build your input dict, and provide it like so: https://github.com/PrefectHQ/prefect/blob/master/src/prefect/client/client.py#L877

ohh sorry I read it wrong 🙂

All good! Want to make sure I'm being clear/helpful here. 🙂

you are! thanks!!~

@Avi A @Zachary Hughes - hope you dont mind me jumping in on this thread - but while splitting the flow into two flows sounds like a quick fix for this specific problem, what if the flow was a sequence of many mapped tasks (i.e. instead of A to C) - (A to Z where Z is the other task … ) ? Shouldn’t prefect’s cache be a bit smarter about not having to retrieve all the previous tasks data ?

Hi @Marwan Sarieddine, absolutely, and that's why I encouraged Avi to open an issue so we can track and handle this behavior! In the meantime, the goal was to get Avi unblocked. 🙂

ah ok gotcha ! thanks for the quick reply