code: def run_ingest_bulk(dataset_size_gb, num_workers):
      ds = make_ds(dataset_size_gb, parallelism=200)
      consumers = [
          ConsumingActor.options(
              scheduling_strategy="SPREAD", num_cpus=0.5
          ).remote(i)
          for i in range(num_workers)
      ]
      ds = ds.map_batches(lambda df: df * 2, batch_format="pandas")
      splits = ds.split(
          num_workers, equal=True, locality_hints=consumers
      )
      future = [
          consumers[i].consume.remote(s) for i, s in enumerate(splits)
      ]
      ray.get(future)
  
file_path: ray/release/nightly_tests/dataset/data_ingest_benchmark.py
imports: import numpy
  import sys
  import time
  import argparse
  from benchmark import Benchmark
  import ray
  from ray.data import Dataset
  from ray.data import DatasetPipeline
  import pandas
  import torch
label_desc: Use image node_id 4 for calling a global function with example usage: run_ingest_bulk(dataset_size_gb, num_workers) without return types