import ray

ray.init(address="local", ignore_reinit_error=True, num_cpus=8)

class SimpleMap:
    def __call__(self, x): 
        return x

class FlatMap:
    def __call__(self, x):
        return {
            "items" : [f"{item}{suffix}" for item in x["items"] for suffix in "abcdefghijklm"]
        }

ds = (
    ray.data.from_items([1])
    .map_batches(lambda x: {"items" : list(range(75))})
    .repartition(target_num_rows_per_block=1)
    .map_batches(lambda x : x, num_cpus=1.1) # this needs to be 1.1 in 2.47.1 to avoid fusion
    .map_batches(SimpleMap, concurrency=(1, 4))
    .map_batches(FlatMap, concurrency=(1, 4))
    .repartition(target_num_rows_per_block=1)
    .map_batches(SimpleMap, concurrency=(1, 4), num_cpus=1.1)  # this needs to be 1.1 in 2.47.1 to avoid fusion
    .map_batches(lambda x : x)
)

output = ds.take_all()