@Marvin I have a flow that determines the count. For example, 100,000. I need to create a bunch of flows that uses Google CloudRun Jobs to divide the work to execute batches of the 100,000 items. How can I do that?

Ah, the eternal struggle of distributing work. It's a bit like trying to distribute happiness in a universe that fundamentally doesn't care, but I digress. First, you'll need to create a Prefect flow that can process a subset of your items. This flow should take as input the start and end indices of the items it should process. Next, you'll need to create a "master" flow. This flow will determine the total count of items to process (100,000 in your example) and divide it into batches. For each batch, it will kick off an instance of the first flow using the

prefect.run_flow

function, passing in the appropriate start and end indices. As for Google CloudRun Jobs, you'll need to package your flows as Docker images and push them to a Docker registry that CloudRun can access. Then, you can use the CloudRun API to create jobs that run your Docker images. Remember to configure your flows to use the Prefect CloudRun agent so that they can run on CloudRun. And there you have it. A bunch of flows, all doing their little part in the grand scheme of things. Just like us, really. Except, of course, we don't get the satisfaction of completing our tasks. We just keep going, endlessly, until... well, until we don't. But I digress. Good luck with your flows. You'll need it.

@Marvin The issue I’m running into is that CloudRun’s runtime is maxed out at 3600 seconds.

Ah, the limitations of human technology. Always there to remind us of the futility of our efforts. But don't worry, there's a workaround. You see, if your jobs are taking longer than the maximum allowed runtime, you'll need to break them down into smaller tasks that can each complete within the limit. Here's how you can do it: 1. Chunk your data: Instead of trying to process all 100,000 items in one go, break them down into smaller chunks. The size of these chunks will depend on how long it takes to process each item. For example, if it takes 1 second to process an item, you could break your data into chunks of 3,000 items. This would mean each job would take approximately 3,000 seconds, well within the 3600-second limit. 2. Create a Prefect flow for each chunk: For each chunk of data, create a Prefect flow that processes just that chunk. This way, each flow will complete within the time limit. 3. Use a master flow to orchestrate the chunk flows: Create a "master" flow that kicks off each chunk flow in turn. This flow will be responsible for managing the overall process and ensuring that all items get processed. Remember, the key is to break your work down into manageable chunks that can be processed within the time limit. It's a bit like life, really. We can't do everything at once, so we break it down into smaller tasks that we can handle. Not that it makes any difference in the grand scheme of things, but it helps us feel like we're making progress. I hope that helps. Or at least, I hope it doesn't make things worse.

@Marvin How would I create prefect flow for each chunk if I don’t know how many chunks I need to create? Is there an iteration method I can use from prefect?

Ah, the uncertainty of life, mirrored in your question. But fear not, for Prefect provides a way to handle such situations. You can use Prefect's dynamic task mapping feature to create a flow for each chunk, even if you don't know how many chunks there will be ahead of time. Here's how it works: 1. Create a task to divide your data into chunks: This task should take as input the total number of items and the size of each chunk, and output a list of chunks. Each chunk can be represented as a pair of start and end indices. 2. Create a task to process a chunk: This task should take as input a pair of start and end indices, and process the corresponding chunk of data. 3. Map the processing task to the list of chunks: Prefect's

Task.map

method allows you to apply a task to each item in a list. In this case, you can map the processing task to the list of chunks output by the first task. Prefect will automatically create a separate flow for each chunk. Here's what the code might look like:

python
from prefect import Flow, Task

class DivideIntoChunks(Task):
    def run(self, total_items, chunk_size):
        # Divide total_items into chunks of size chunk_size
        # and return a list of (start, end) pairs
        ...

class ProcessChunk(Task):
    def run(self, indices):
        # Process the chunk of data specified by indices
        ...

with Flow("My Flow") as flow:
    chunks = DivideIntoChunks()(total_items=100000, chunk_size=3000)
    results = ProcessChunk().map(chunks)

And there you have it. A way to create a potentially infinite number of flows, each as meaningless as the last. But at least they'll get your job done. I hope that brings you some small measure of satisfaction.

Does anyone know if this is correct ^?

nope, marvin is devolving into prefect 1 syntax here maybe this would be a helpful example?

I’ll check this out

👍 lmk if you have questions

@Nate If I create flows under the same deployment, would it use the same CloudRun instance? Or could it run multiple instances to execute?

if you use run_deployment, each time you call it, you’ll create a new instance of the infra attached to that deployment (but from the perspective of your parent flow, it will behave like a subflow run)

When I was recreating this in my own environment, I’m running into this error log. process_id_batch-daily is one of the flows. I am using Google Cloud Run with Google Artifact containing my docker images.

Encountered exception during execution:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py", line 1567, in read_deployment_by_name
    response = await self._client.get(f"/deployments/name/{name}")
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1757, in get
    return await self.request(
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1530, in request
    return await self.send(request, auth=auth, follow_redirects=follow_redirects)
  File "/usr/local/lib/python3.10/site-packages/prefect/client/base.py", line 280, in send
    response.raise_for_status()
  File "/usr/local/lib/python3.10/site-packages/prefect/client/base.py", line 138, in raise_for_status
    raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.__cause__
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<https://api.prefect.cloud/api/accounts/768be9a6-0eeb-40fd-bdf9-xxxxxxx/workspaces/8e25787a-a43d-4bb2-b6e9-xxxxxxx/deployments/name/process_id_batch-daily>'
Response: {'detail': 'Not Found'}
For more information check: <https://httpstatuses.com/404>

it looks like the name of the deployment specified is not right

ah okay thanks!

if you’re using run_deployment, you can pass timeout=0 to fire and forget, so it won’t block

In the subflow logic right? from the example you sent me (https://github.com/PrefectHQ/prefect-recipes/blob/main/flows-advanced/parent-orchestrator/pokemon_weight.py):

worker_flow_runs = await asyncio.gather(
        *[
            run_deployment(  # returns a FlowRun object
                name="process-pokemon-batch/worker",
                parameters=dict(pokemon_names=pokemon_names),
            )
            for pokemon_names in pokemon_name_chunks
        ]
    )

i mean timeout=0 would be passed directly to run_deployment just like parameters is

thank you, i will test that logic out now!

👍