https://prefect.io logo
Join Slack
Powered by
Hi, how do I run multiple tasks? I have a list of ...
# ask-community
b
Hi, how do I run multiple tasks? I have a list of ids from my back end that I need to rerun the tasks in prefect, but seems like i need to do it one by one, is there a way to pass that list and it will create all the tasks and run them?
k
Hi @Bruno Centeno, are these tasks part of different Flows?
b
just one flow
k
Are these tasks that failed and you’re trying to restart the Flow from the point of failure?
b
@Kevin Kho well i have had bth cases, where i want to rerun a failed task or when i want to create a new one
k
If you’re trying to restart a Flow from the point of failure, the easiest way would be to click the Restart button in the UI. I don’t think you can run an individual task as you would need to submit the Flow by creating a Flow run, so you would need to run the Flow again. If you want a mechanism to skip tasks that have already run successfully before, I think you can utilize caching .
b
If you’re trying to restart a Flow from the point of failure, the easiest way would be to click the Restart button in the UI
yes that’s what im doing, but i have several tasks (169) and i dont want to do them one by one. On the other hand I have other 130 that run correctly but my back end had a bug so they didnt do what they were suppose to do and i want to run them again now that i changed that part of my back end(or create new ones with those parameters) is there a way to do multiple tasks at the same time from the UI (of the same flow)
k
If you click the restart here on the top right on the Flow run page (not on the task run page), then it would re-run from the point where the Flow left off and all subsequent tasks failed tasks. Does it only run one task for you?
For the ones that changed, you need to start a new Flow run (and re-register) for sure because the metadata of the Flow has changed. In that case you would run a new Flow entirely. But if you have caching enabled on the previous tasks, you wouldn’t have to run them again (in the future).
b
There is something i am not understanding, how to re run all failed runs? I dont see that button you showed upthere, only when i enter into one of them
maybe i was not using the right lingo, “task” i am sorry i think it was “run”
k
Ohhh ok. I understand what you are saying now. You have 169 Flow runs you want to re-run from the point of failure? So for the ones where your code changed, you still need to register and start those Flow runs again to take effect.
b
yes, sorry that’s what i want to do from the picture above select all of them and make them start again
k
So for this, you need to use the GraphQL API client. One second let me make a sample for you.
👍 1
Ok so there’s two parts to this. First is you can query the flow ids you need with:
Copy code
query{
  flow_run(where: {state: {_eq: "Failed"}}) {
    name
    id
    state
    flow {
      id
      name
      project {
        name
      }
    }
    }
  }
And then you can restart them by changing them to State = Scheduled like this:
Copy code
mutation {
  set_flow_run_states(input: {states: [{
      flow_run_id: "d2d9e33f-ea92-4a91-8971-8e80d0549991", 
      state: "{\"type\": \"Scheduled\", \"message\": \"<<your message>>\"}"}]},) {
    states {
      id
      status
    }
  }
}
So you can try this together with a Python script. 
prefect.client.client.Client
has the 
graphql
method so you can use that to run the queries. Here is example syntax . You’re gonna change the query inside. And then you can loop over the 
flow_run_ids
you want to restart and plug them into this mutation and run the graphQL call.
For the ones you changed the backend of, you can’t do this as this will run the archived version (and error), you need to start new flow runs for those.
And then there’s also one more caveat that the graphql api returns 100 results so you need to run the query with an offset to get the others later.
b
Can i use the Interactive API? to create the new ones i should do something like this?:
Copy code
mutation {
  createFlowRun(input: {
      parameters {
      	id: ["143b0976-0261-4aa6-8082-af5051190610", 
        	"8a597bd8-b98a-4cd6-8908-ddedf326550b",
        	"fc35c91f-49a3-40cc-b4fa-7c3e9c6531d1"]
    	}
  }
}
I want to pass the list of ids to create multiple runs, the picture is like i do it manually
k
This API endpoint only takes in one flow run. It’s easier to use the Python API.
Copy code
from prefect.client.client import Client
client = Client()

for id in ids:
    client.create_flow_run(id)
Something like this will be easier.
b
oh i see
i will take a look at that thank you
👍 1
4 Views
Open in Slack
PreviousNext
Powered by