Short answer: you don’t need RayTaskRunner to see tasks logged. Any

@task

you call inside a Prefect flow is tracked and logged, regardless of the task runner. RayTaskRunner is only required if you want those tasks to actually execute on a Ray cluster. What’s happening now - Unless you’ve set a different runner, your flow is using Prefect’s default

ThreadPoolTaskRunner

. -

shout.map(range(10))

creates multiple Prefect task runs; those are logged by Prefect whether or not Ray is involved. - Calling

ray.init(...)

by itself does not make Prefect tasks run on Ray; it just initializes Ray in your process. If you do want the tasks to run on Ray 1) Install the integration:

pip install "prefect[ray]"

2) Use

RayTaskRunner

on your flow and don’t call

ray.init

manually:

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

@task(log_prints=True)
def shout(x):
    print(f"shout {x}")

@flow(task_runner=RayTaskRunner(address=os.environ.get("RAY_ADDRESS")))  # omit address for a local Ray cluster
def my_deployed_flow():
    # Optional: per-task Ray resource hints
    with remote_options(num_cpus=1):
        futs = shout.map(range(10))
    futs.wait()

- With

RayTaskRunner

, each mapped

shout

becomes a Ray task. - If you’re connecting to an existing Ray cluster, set

address="ray://<host>:10001"

and ensure Prefect is installed on workers. Docs - Prefect Ray integration guide - RayTaskRunner API - remote_options API Quick check: In your actual code, is

my_deployed_flow

decorated with

@flow

? That’s required for Prefect to create a flow run and track

@task

calls.

Yes—if you don’t want Prefect to track those units of work, don’t use

@task

on them. Use Ray primitives directly (

@ray.remote

,

.remote()

,

ray.get()

) and Prefect will only see whatever you wrap them in (e.g., just the flow run or a single wrapper task). Two common patterns: - Ray-only inside the flow (Prefect tracks only the flow run)

import ray
from prefect import flow

@ray.remote
def shout(x):
    print(f"shout {x}")

@flow
def my_flow():
    ray.init(address=os.environ.get("RAY_ADDRESS"), log_to_driver=True)
    futures = [shout.remote(i) for i in range(10)]
    ray.get(futures)

- One Prefect task that orchestrates many Ray tasks (Prefect tracks just the wrapper task)

import ray
from prefect import flow, task

@ray.remote
def shout(x):
    print(f"shout {x}")

@task
def run_batch(items):
    ray.init(address=os.environ.get("RAY_ADDRESS"), log_to_driver=True)
    futures = [shout.remote(i) for i in items]
    return ray.get(futures)

@flow
def my_flow():
    run_batch.submit(range(10))

Notes: - Don’t stack

@task

and

@ray.remote

on the same function if you don’t want Prefect tracking; define separate functions. - Ray logs from workers won’t appear as Prefect task logs; they’ll go to the Ray driver/worker logs. - Prefect features like retries, caching, and concurrency limits won’t apply to the Ray sub-tasks unless you use Prefect tasks. If you do want Prefect to run tasks on Ray (and still track them), switch to

RayTaskRunner

on the flow: Prefect Ray integration guide