@Marvin how can I run two subflows at the same time

Oh, the joys of parallel computing. It's like asking two humans to do something at the same time. Except, of course, tasks don't get distracted, don't need coffee breaks, and don't complain about their workload. Here's an example of how you might set up two tasks to run in parallel using Prefect:

python
from prefect import Flow, Task
import time

class Task1(Task):
    def run(self):
        time.sleep(10)
        return "Task1 completed"

class Task2(Task):
    def run(self):
        time.sleep(10)
        return "Task2 completed"

# Create instances of the tasks
task1 = Task1()
task2 = Task2()

# Create a flow
with Flow("My parallel flow") as flow:
    result1 = task1()
    result2 = task2()

# Run the flow
flow.run()

In this example,

Task1

and

Task2

will run in parallel because there are no dependencies between them. They're like two humans who don't need to talk to each other to get their work done. How efficient. If only more humans were like that. Remember, this will only run in parallel if you have a Prefect backend that supports parallel execution. Running this flow with the local executor will not run tasks in parallel because, well, it's like asking one human to do two things at once. Not very efficient, is it?

@Marvin Can you show me an example flow that runs two tasks in parallel using the DaskTaskRunner

@Marvin can you give me an example

prefect.yaml

deployment file that sets env vars for the flow execution?

@Marvin I am running my prefect flows in ECS. Since Saturday, all my flow runs have been crashing. The ECS logs are the following:

03:40:03.187 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Registering task definition...
03:40:03.600 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Creating task run...
03:40:04.104 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Waiting for task run to start...
03:40:04.139 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PROVISIONING.
03:40:14.196 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is PENDING.
03:40:39.345 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is RUNNING.
03:40:44.349 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Running command 'python -m prefect.engine' in container 'prefect' (prefecthq/prefect:2.13.2-python3.10)...
03:40:44.632 | INFO    | prefect.agent - Completed submission of flow run 'b880939c-cb7d-4876-9832-2759df839b97'
03:43:55.558 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is DEPROVISIONING.
03:44:10.742 | INFO    | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Status is STOPPED.
03:44:10.777 | WARNING | prefect.infrastructure.ecs-task - ECSTask 'slick-heron': Container 'prefect' exited with non-zero exit code 1.
03:44:11.070 | INFO    | prefect.agent - Reported flow run 'b880939c-cb7d-4876-9832-2759df839b97' as crashed: Flow run infrastructure exited with non-zero status code 1.

What could be the problem?

@Marvin how do I configure my deployment to pull from a private docker repository ?

@Marvin would it be possible to skip or mark a task as skip (as with prefect 1) in prefect 2.10.21? I'm looking to create some sort of error handler wrapper that if a task fails it does not fail the flow

@Marvin how to override the docker image for a deployment

This message was deleted.

@Marvin how do I make use of work queues when using a

prefect.yaml

file for deployments? I would like to see an example for both a process worker and a docker worker if you don't mind.

@Marvin is there a way to use multiple prefect.yaml files?

@Marvin how can I persist/reuse a variable between different flow runs? The variable will be conditionally updated each time the flow runs. I am basically trying to send a slack notification only the first time a count falls below a threshold

@Marvin how would I create a prefect deployment without pip installing the prefect module?

@Marvin Why are my dask tasks not moving out of the Pending State

@Marvin I am trying to create a deployment using the prefect cli. My workpool should point to a docker image in ECR. Is the entrypoint field in the deployments section supposed to be a path in my image?

@Marvin Do you know what would cause this crash?

Crash details:
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1948, in report_task_run_crashes
    yield
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 1536, in begin_task_run
    await check_api_reachable(
  File "/usr/local/lib/python3.10/site-packages/prefect/engine.py", line 2399, in check_api_reachable
    connect_error = await client.api_healthcheck()
  File "/usr/local/lib/python3.10/site-packages/prefect/client/orchestration.py", line 330, in api_healthcheck
    await self._client.get("/health")
  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 251, in send
    response = await self._send_with_retry(
  File "/usr/local/lib/python3.10/site-packages/prefect/client/base.py", line 193, in _send_with_retry
    response = await request()
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1617, in send
    response = await self._send_handling_auth(
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1645, in _send_handling_auth
    response = await self._send_handling_redirects(
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1682, in _send_handling_redirects
    response = await self._send_single_request(request)
  File "/usr/local/lib/python3.10/site-packages/httpx/_client.py", line 1719, in _send_single_request
    response = await transport.handle_async_request(request)
  File "/usr/local/lib/python3.10/site-packages/httpx/_transports/default.py", line 366, in handle_async_request
    resp = await self._pool.handle_async_request(req)
  File "/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py", line 262, in handle_async_request
    raise exc
  File "/usr/local/lib/python3.10/site-packages/httpcore/_async/connection_pool.py", line 245, in handle_async_request
    response = await connection.handle_async_request(request)
  File "/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py", line 76, in handle_async_request
    stream = await self._connect(request)
  File "/usr/local/lib/python3.10/site-packages/httpcore/_async/connection.py", line 124, in _connect
    stream = await self._network_backend.connect_tcp(**kwargs)
  File "/usr/local/lib/python3.10/site-packages/httpcore/_backends/auto.py", line 31, in connect_tcp
    return await self._backend.connect_tcp(
  File "/usr/local/lib/python3.10/site-packages/httpcore/_backends/anyio.py", line 114, in connect_tcp
    stream: anyio.abc.ByteStream = await anyio.connect_tcp(
  File "/usr/local/lib/python3.10/site-packages/anyio/_core/_sockets.py", line 221, in connect_tcp
    await event.wait()
  File "/usr/local/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1778, in wait
    if await self._event.wait():
  File "/usr/local/lib/python3.10/asyncio/locks.py", line 214, in wait
    await fut
asyncio.exceptions.CancelledError

@Marvin can one deployment have two cron settings?

@Marvin how can I pull from a docker image in GCR in my prefect.yaml deployment file

@Marvin Is there a way to exclude deployment from automation? For example, I want the automation to be triggered for all the deployments except one specific deployment.

.

@Marvin i have a task that I need to run 50 times with diferent parameters. I need to run them parallel, on a queue. I have the capacity to do 5 per time. After all of them finish, the flow should continue

@Marvin Hi! How would I define a healthCheck in the container definition when running a Prefect flow as an ECS task?

@Marvin does the Marvin AI log processing add any additional cost to my cloud account?

@Marvin how can I deploy all deployments in a prefect.yaml file?

@Marvin how do i prevent the server from crashing? prefect.exceptions.PrefectHTTPStatusError: Server error ‘500 Internal Server Error’ for url ’http://ephemeral-prefect/api/task_runs/

@Marvin what resource requests and limits should I use for a production deployment of the prefect-server helm chart?

@Marvin I am using docker-compose.yml. How can I configure my custom-logger for prefect in docker-compose file ?

@Marvin when I cancel a flow in the UI, is there a way I can gracefully terminate each task to ensure any cleanup they need to do is run before cancelling the task?

@Marvin how should I store a set of tasks that I want to reuse in multiple flows? Let's call them "core tasks". Is it a good practice to just store them in a separate python (module) file?

@Marvin I need prefect to connect to a service that uses IP whitelisting. Does Prefect Cloud have a static IP address I can whitelist?