[Data] map_batches fails with asyncio.run() in func and chaining with acync actor due to event loop conflict

What happened + What you expected to happen

When I use the map_batches method of Ray Data, chaining a function containing asyncio.run with an async actor, the program raises an error ”RuntimeError: asyncio.run() cannot be called from a running event loop“. Both functions work successfully when used independently, but fail when chained together.

We use an async function because the process may involve network I/O and does not require storing state, so we used an async function instead of chaining two async actors. Does Ray Data have plans to support passing async functions in map_batches?

Versions / Dependencies

ray 2.50.0
python 3.12.9

Reproduction script

import ray.data
import asyncio


def load_data(batch):
    def _fake_load(p):
        return p

    async def _async_load_all():
        return await asyncio.gather(
            *(asyncio.to_thread(_fake_load, path) for path in batch["item"])
        )

    batch["data"] = asyncio.run(_async_load_all())
    return batch


class AsyncActor:
    async def __call__(self, inp):
        await asyncio.sleep(0.1)  # Simulate async operations
        return inp


if __name__ == "__main__":
    # succeed with load_data func
    # ray.data.from_items(["a", "b"]).map_batches(load_data).show()

    # succeed with AsyncActor actor
    # ray.data.from_items(["a", "b"]).map_batches(
    #    AsyncActor, concurrency=1
    # ).show()

    # expect succeed but failed
    ray.data.from_items(["a", "b"]).map_batches(load_data).map_batches(
        AsyncActor, concurrency=1
    ).show()

Issue Severity

Medium: It is a significant difficulty but I can work around it.

Start by running the reproduction script against Ray 2.50.0 and Python 3.12.9, comparing the two independent map_batches calls with the chained call. Trace the map_batches execution paths for the synchronous function and async actor; done means the chained example no longer raises the event-loop conflict, or the supported limitation and workaround are documented.