I ran a sample and there were no problems occurred.

import ray
from typing import Callable


class Op12(Callable):
    def __call__(self, record) -> str:
        return {"value": "op1"*1000000}


ray.init(ignore_reinit_error=True, runtime_env={
         "working_dir": './', "excludes": ["CC*", "part*", "PMC*"]})
ray.data.from_items(range(1, 10000)).map(Op12,  concurrency=10).take_all()

@Jay-ju you are using local ray address; you need to test this with ray cluster address

@Jay-ju you are using local ray address; you need to test this with ray cluster address

i am using ray cluster mode， I actually don't quite get what you mean.

@Jay-ju Sorry If I was unclear; in your code, address = None for the ray.init() so; I thought it was running as a local cluster in same machine where the code is; in that case that runtime env for the working dir is not needed as local cluster has access to those folder; Issues was occurred when the ray cluster was setup in different machine and then actual code is another machine. I have a distributed cluster setup in azure and the same working dir passing as runtime env works for ray core; but it wasn't working for those ray.data; if it is not too much trouble; could you please try with this part of code:

It is possible that it is trying to calculate metrics for this pandas_refs from ray.data but may not be calculating for that above part of your code.

import pandas as pd 
data= { 'A': [1, 2, 3, 4, 5], 'B': ['a', 'b', 'c', 'd', 'e'] }
df = pd.DataFrame(data)

obj1 = ray.put(df )
obj2 = ray.put(df)
ray_objs = [obj1, obj2]

dataset = ray.data.from_pandas_refs(ray_objs)

for bs in datset.iter_batches(prefetch_batches=0, batch_size=1000, batch_format="pandas", drop_last=False):
    pass