prefect.runtime

receive_input

send_input

receive_input

receive_input

send_input

python
from prefect import flow
from prefect_runtime import RunInput

@flow
def sender_flow():
    # Sending a message or data to the receiver flow
    RunInput.send_input(flow_run_id='receiver_flow_run_id', input_data='Hello from Sender Flow!')

python
from prefect import flow
from prefect_runtime import receive_input

@flow
def receiver_flow():
    # Create an iterator to receive inputs
    receiver = receive_input()
    
    # Loop to handle incoming inputs
    for run_input in receiver:
        print(f"Received input: {run_input.input_data}")
        if run_input.input_data == 'EXIT_SIGNAL':
            break  # Exit loop if exit signal received

sender_flow

receiver_flow

receiver_flow

python
import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@flow(task_runner=ConcurrentTaskRunner())
def generate_subtasks():
    subtasks = [generate_random_number.submit() for _ in range(100)]
    # Collect the results of the subtasks
    subtask_results = [subtask.result() for subtask in subtasks]
    return aggregate_numbers(subtask_results)

@flow(task_runner=ConcurrentTaskRunner())
def main_flow():
    # Generate main tasks, each producing its own subtasks
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
    # Collect results of main tasks
    main_task_results = [main_task.result() for main_task in main_tasks]
    total_sum = aggregate_numbers(main_task_results)
    print(f"Total sum of all random numbers: {total_sum}")

if __name__ == "__main__":
    main_flow()

generate_subtasks

ConcurrentTaskRunner

11:25:16.576 | INFO    | prefect.engine - Created flow run 'noisy-heron' for flow 'main-flow'
11:25:16.578 | INFO    | prefect.engine - View at <http://prefect-server.prefect.svc.cluster.local:4200/runs/flow-run/c4e18255-8700-40ab-8c97-96fd58a35374>
11:25:16.736 | ERROR   | Flow run 'noisy-heron' - Encountered exception during execution: AttributeError("'Flow' object has no attribute 'submit'")
Traceback (most recent call last):
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 655, in run_context
    yield self
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 699, in run_flow_sync
    engine.call_flow_fn()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 678, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/utilities/callables.py", line 206, in call_with_parameters
    return fn(*args, **kwargs)
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in main_flow
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in <listcomp>
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
AttributeError: 'Flow' object has no attribute 'submit'
11:25:16.822 | ERROR   | Flow run 'noisy-heron' - Finished in state Failed("Flow run encountered an exception: AttributeError: 'Flow' object has no attribute 'submit'")
Traceback (most recent call last):
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 30, in <module>
    main_flow()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flows.py", line 1355, in __call__
    return run_flow(
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 821, in run_flow
    return run_flow_sync(**kwargs)
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 701, in run_flow_sync
    return engine.state if return_type == "state" else engine.result()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 255, in result
    raise self._raised
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 655, in run_context
    yield self
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 699, in run_flow_sync
    engine.call_flow_fn()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 678, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/utilities/callables.py", line 206, in call_with_parameters
    return fn(*args, **kwargs)
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in main_flow
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in <listcomp>
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
AttributeError: 'Flow' object has no attribute 'submit'

running your snippet it returned the following error:

return run_flow_sync(**kwargs)
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 701, in run_flow_sync
    return engine.state if return_type == "state" else engine.result()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 255, in result
    raise self._raised
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 655, in run_context
    yield self
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 699, in run_flow_sync
    engine.call_flow_fn()
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/flow_engine.py", line 678, in call_flow_fn
    result = call_with_parameters(self.flow.fn, self.parameters)
  File "/home/huguntu/anaconda3/envs/prefect-py39/lib/python3.9/site-packages/prefect/utilities/callables.py", line 206, in call_with_parameters
    return fn(*args, **kwargs)
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in main_flow
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in <listcomp>
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
AttributeError: 'Flow' object has no attribute 'submit'

Your solution outputs the following error when executed:

return fn(*args, **kwargs)
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in main_flow
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 23, in <listcomp>
    main_tasks = [generate_subtasks.submit() for _ in range(100)]
AttributeError: 'Flow' object has no attribute 'submit'

submit

task_runners

python
import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@flow(task_runner=ConcurrentTaskRunner())
def generate_subtasks():
    subtasks = [generate_random_number() for _ in range(100)]
    subtask_results = [subtask.result() for subtask in subtasks]
    return aggregate_numbers(subtask_results)

@flow(task_runner=ConcurrentTaskRunner())
def main_flow():
    # Generate main tasks, each producing its own subtasks
    main_task_results = [generate_subtasks() for _ in range(100)]
    total_sum = aggregate_numbers(main_task_results).result()
    print(f"Total sum of all random numbers: {total_sum}")

if __name__ == "__main__":
    main_flow()

generate_subtasks()

main_flow

.submit()

.result()

ConcurrentTaskRunner

main_flow

generate_subtasks

File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 16, in generate_subtasks
    subtask_results = [subtask.result() for subtask in subtasks]
  File "/home/huguntu/jungleai/miguel.farrajota/prefect-test-code/nested_graphs.py", line 16, in <listcomp>
    subtask_results = [subtask.result() for subtask in subtasks]
AttributeError: 'int' object has no attribute 'result'

.result()

.result()

python
import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@flow(task_runner=ConcurrentTaskRunner())
def generate_subtasks():
    # Run 100 tasks to generate random numbers
    subtask_results = [generate_random_number() for _ in range(100)]
    # Aggregate the results
    return aggregate_numbers(subtask_results)

@flow(task_runner=ConcurrentTaskRunner())
def main_flow():
    # Run 100 main tasks, each executing its own subtasks
    main_task_results = [generate_subtasks() for _ in range(100)]
    # Aggregate all the results
    total = aggregate_numbers(main_task_results)
    print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

.result()

generate_subtasks

main_flow

ConcurrentTaskRunner

python
import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@flow(task_runner=ConcurrentTaskRunner())
def main_flow():
    # Generate a list of tasks for generating random numbers
    random_number_tasks = [generate_random_number.submit() for _ in range(10000)]
    
    # Retrieve results and aggregate them
    random_number_results = [task.result() for task in random_number_tasks]
    total = aggregate_numbers(random_number_results)
    
    # Output the total sum
    print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

main_flow

generate_random_number

.submit()

ConcurrentTaskRunner

python
import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@flow(task_runner=ConcurrentTaskRunner())
def main_flow():
    # Create 100 tasks, each of which generates another 100 tasks
    all_random_number_tasks = [
        [generate_random_number.submit() for _ in range(100)] for _ in range(100)
    ]
    
    # Flatten the list of lists into a single list
    all_random_number_results = []
    for task_list in all_random_number_tasks:
        for task in task_list:
            all_random_number_results.append(task.result())

    # Aggregate the results of all tasks
    total = aggregate_numbers(all_random_number_results)
    
    # Output the total sum
    print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

main_flow

.submit()

ConcurrentTaskRunner

aggregate_numbers

be converted to 'SchemaSerializer'")
15:13:14.682 | ERROR   | Flow run 'thoughtful-dog' - Finished in state Failed("Flow run encountered an exception: TypeError: 'MockValSer' object cannot be converted to 'SchemaSerializer'")
Traceback (most recent call last):

import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@task
def generate_subtasks():
    subtask_results = [generate_random_number.submit() for _ in range(10)]
    # Aggregate the results
    return aggregate_numbers([subtask.result() for subtask in subtask_results])

#@flow(task_runner=ConcurrentTaskRunner())
@flow(
    task_runner=RayTaskRunner(
        address="<ray://kuberay-cluster-head-svc.ray.svc.cluster.local:10001>",
        init_kwargs={"runtime_env": {"pip": ["prefect[ray]"]}},
    )
)
def main_flow():
    with remote_options(num_cpus=4):
        all_random_number_tasks = []
        for _ in range(10):
            all_random_number_tasks.append(generate_subtasks.submit())

        results = [task.result() for task in all_random_number_tasks]

        # Aggregate the results of all tasks
        total = aggregate_numbers(results)

        # Output the total sum
        print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

different subject: Whenever I try to run the following code with a ray task runner, it returns me this error:

be converted to 'SchemaSerializer'")
15:13:14.682 | ERROR   | Flow run 'thoughtful-dog' - Finished in state Failed("Flow run encountered an exception: TypeError: 'MockValSer' object cannot be converted to 'SchemaSerializer'")
Traceback (most recent call last):

be converted to 'SchemaSerializer'")
15:13:14.682 | ERROR   | Flow run 'thoughtful-dog' - Finished in state Failed("Flow run encountered an exception: TypeError: 'MockValSer' object cannot be converted to 'SchemaSerializer'")
Traceback (most recent call last):

RayTaskRunner

'MockValSer' object cannot be converted to 'SchemaSerializer'

prefect-ray

prefect-ray

shell
   pip install prefect-ray

prefect-ray

import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@task
def generate_subtasks():
    subtask_results = [generate_random_number.submit() for _ in range(10)]
    # Aggregate the results
    return aggregate_numbers([subtask.result() for subtask in subtask_results])

@flow(
    task_runner=RayTaskRunner(
        address="<ray://kuberay-cluster-head-svc.ray.svc.cluster.local:10001>",
        init_kwargs={"runtime_env": {"pip": ["prefect[ray]"]}},
    )
)
def main_flow():
    with remote_options(num_cpus=4):
        all_random_number_tasks = []
        for _ in range(10):
            all_random_number_tasks.append(generate_subtasks.submit())

        results = [task.result() for task in all_random_number_tasks]

        # Aggregate the results of all tasks
        total = aggregate_numbers(results)

        # Output the total sum
        print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

The

prefect-ray

package is already installed ad up-to-date. If it helps you to diagnose the issue, here is the code I'm running that generated the previous error:

prefect-ray

import random
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@task
def generate_subtasks():
    subtask_results = [generate_random_number.submit() for _ in range(10)]
    return aggregate_numbers([subtask.result() for subtask in subtask_results])

@flow(
    task_runner=RayTaskRunner(
        address="<ray://kuberay-cluster-head-svc.ray.svc.cluster.local:10001>",
        init_kwargs={"runtime_env": {"pip": ["prefect[ray]"]}},
    )
)
def main_flow():
    with remote_options(num_cpus=4):
        all_random_number_tasks = []
        for _ in range(10):
            all_random_number_tasks.append(generate_subtasks.submit())

        results = [task.result() for task in all_random_number_tasks]
        total = aggregate_numbers(results)
        print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

prefect-ray

RayTaskRunner

address

runtime_env

pip

generate_random_number

aggregate_numbers

.submit()

generate_subtasks

remote_options

num_cpus=4

python
import random
from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def generate_random_number():
    return random.randint(1, 5)

@task
def aggregate_numbers(numbers):
    return sum(numbers)

@task
def generate_subtasks():
    subtask_results = [generate_random_number.submit() for _ in range(10)]
    results = [result.result() for result in subtask_results]
    return aggregate_numbers(results)

@flow(
    task_runner=RayTaskRunner(
        address="<ray://kuberay-cluster-head-svc.ray.svc.cluster.local:10001>",
        init_kwargs={"runtime_env": {"pip": ["prefect[ray]"]}},
    )
)
def main_flow():
    # Ensure num_cpus aligns with Ray's available cluster resources
    with remote_options(num_cpus=4):
        all_random_number_tasks = [generate_subtasks.submit() for _ in range(10)]
        
        # Collect the results from Ray tasks
        results = [task.result() for task in all_random_number_tasks]
        total = aggregate_numbers(results)
        print(f"Total sum of all random numbers: {total}")

if __name__ == "__main__":
    main_flow()

num_cpus

prefect.yaml

deployment_id

prefect deployment ls

python
     import httpx

     deployment_id = "your-deployment-id"
     parameters = {"param1": value1, "param2": value2}

     response = <http://httpx.post|httpx.post>(f"<https://api.prefect.io/deployments/{deployment_id}/flows/run>", json={"parameters": parameters})

prefect variable set <name> <value>

prefect variable get <name>

prefect variable unset <name>

prefect variable ls

prefect variable inspect <name>

flow.run()

fire_and_forget

python
from prefect import flow, task

@task
def start_another_flow(flow_fn, *args, **kwargs):
    flow_fn(*args, **kwargs)

@flow
def sub_flow_example(param):
    print(f"Running sub-flow with parameter: {param}")

@flow
def main_flow():
    # Initiating subflows without waiting for them to complete
    for i in range(5):
        start_another_flow(sub_flow_example, param=i)

    # Immediately proceed to terminate the main flow
    print("Main flow is terminating after starting sub-flows.")

if __name__ == "__main__":
    main_flow()

start_another_flow

visualize()

visualize()

visualize()

visualize()

python
from prefect import flow

@flow
def simple_flow():
    # Your flow logic here

# Visualize the flow
simple_flow.visualize()

visualize()