Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

<@ULVA73B9P> its possible to have concurrent execution in a pipeline

:white_check_mark: thought for 13.0 seconds

Yes — in *Prefect 3.x*, concurrent execution in a pipeline/flow is definitely possible.

A few common ways:

*1. Run multiple tasks at the same time*
If tasks don’t depend on each other, Prefect can submit them concurrently:

```python
from prefect import flow, task

@task
def step_1():
    return "a"

@task
def step_2():
    return "b"

@flow
def my_flow():
    a = step_1.submit()
    b = step_2.submit()

    print(a.result(), b.result())

my_flow()
```

Using `.submit()` schedules tasks for concurrent execution instead of running them one-by-one.

*2. Use `map`-style parallelism*
If you want to run the same task over many inputs:

```python
from prefect import flow, task

@task
def process(x):
    return x * 2

@flow
def my_flow():
    futures = [process.submit(i) for i in range(10)]
    results = [f.result() for f in futures]
    print(results)

my_flow()
```

*3. Use async flows/tasks*
If your work is naturally async (API calls, I/O, etc.), Prefect also supports `async` flows and tasks.

*Important caveat*
Concurrency depends on your *task runner* and execution environment:
- The default runner can run submitted tasks concurrently
- For heavier parallelism, you may want a different task runner like `ThreadPoolTaskRunner` or `ProcessPoolTaskRunner`

Example with a thread pool:

```python
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner

@task
def work(x):
    return x + 1

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def my_flow():
    futures = [work.submit(i) for i in range(10)]
    return [f.result() for f in futures]

my_flow()
```

A couple of related docs:
- <https://docs.prefect.io|Prefect docs>
- <https://docs.prefect.io/v3/how-to-guides/workflows/run-work-concurrently|Run work concurrently>

If you want, I can also show:
- *task-level concurrency within one flow*
- *running multiple flow runs concurrently*
- *limits/throttling concurrency*