@Marvin is there a way to limit how many concurrent tasks get spawned when using concurrenttaskrunner()?

You can limit the number of concurrent tasks when using

ConcurrentTaskRunner

by applying task concurrency limits via task tags. Here's how you can set it up: 1. Add Tags to Tasks: Apply tags to your tasks and then specify a concurrency limit for those tags. 2. Configure Task Concurrency Limits: You can specify the maximum number of concurrent task runs for tasks with a given tag. For more details, you can refer to the official Prefect documentation on task concurrency limits. In your flow, you would do something like this:

python
from prefect import flow, task

@task(tags=["my_tag"])
def my_task():
    # your task logic

@flow(name="My Flow")
def my_flow():
    # your flow logic

if __name__ == "__main__":
    my_flow()

Then, set the concurrency limit for the tag through your deployment configuration or Prefect API:

python
from prefect.client import get_client

client = get_client()
client.set_task_run_concurrency_limit("my_tag", limit=5)

This setup ensures that no more than 5 tasks with the tag my_tag will run concurrently. For more information, check out the global concurrency limits guide.