code: def run_ingest_streaming(
      dataset_size_gb, num_workers, use_gpu, early_stop
  ):
      ds = make_ds(dataset_size_gb)
      resources = {"num_cpus": 0.5}
      if use_gpu:
          resources["num_gpus"] = 0.5
      consumers = [
          ConsumingActor.options(
              scheduling_strategy="SPREAD", **resources
          ).remote(i)
          for i in range(num_workers)
      ]
      locality_hints = ray.get(
          [actor.get_location.remote() for actor in consumers]
      )
      ds = ds.map_batches(lambda df: df * 2, batch_format="pandas")
      splits = ds.streaming_split(
          num_workers, equal=True, locality_hints=locality_hints
      )
      max_bytes_to_read = None
      if early_stop:
          max_bytes_to_read = dataset_size_gb * GiB // num_workers // 2
      # Early stop when we've read half the dataset.
      future = [
          consumers[i].consume.remote(
              s,
              use_gpu,
              max_bytes_to_read,
          )
          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 3 for calling a global function with example usage: run_ingest_streaming(dataset_size_gb, num_workers, use_gpu, early_stop) without return types