class_docstr: Utility class used for Ray Data release tests and benchmarks, which works
  for both local and distributed benchmarking. When run on the nightly release
  test pipeline, the results are written to our internal database, which
  can then be rendered on the dashboard. Usage tips:
  
  A typical workflow would be:
  
  benchmark = Benchmark("benchmark-name")
  
  # set up (such as input read or generation)
  ...
  
  benchmark.run_materialize_ds("case-1", fn_1)
  # Could be Ray Data iterator, Torch DataLoader, TF Dataset...
  benchmark.run_iterate_ds("case-2", dataset)
  benchmark.run_fn("case-3", fn_3)
  
  # Writes a JSON with metrics of the form:
  # {"case-1": {...}, "case-2": {...}, "case-3": {...}}
  benchmark.write_result()
  
  See example usage in ``aggregate_benchmark.py``.
class_docstr_tok: ["Utility","class","used","for","Ray","Data","release","tests","and","benchmarks",",","which","works","for","both","local","and","distributed","benchmarking",".","When","run","on","the","nightly","release","test","pipeline",",","the","results","are","written","to","our","internal","database",",","which","can","then","be","rendered","on","the","dashboard",".","Usage","tips",":","A","typical","workflow","would","be",":","benchmark","=","Benchmark","(","``","benchmark-name","''",")","#","set","up","(","such","as","input","read","or","generation",")","...","benchmark.run_materialize_ds","(","``","case-1","''",",","fn_1",")","#","Could","be","Ray","Data","iterator",",","Torch","DataLoader",",","TF","Dataset","...","benchmark.run_iterate_ds","(","``","case-2","''",",","dataset",")","benchmark.run_fn","(","``","case-3","''",",","fn_3",")","#","Writes","a","JSON","with","metrics","of","the","form",":","#","{","``","case-1","''",":","{","...","}",",","``","case-2","''",":","{","...","}",",","``","case-3","''",":","{","...","}","}","benchmark.write_result","(",")","See","example","usage","in","``","aggregate_benchmark.py","``","."]
docstr: Run a benchmark on materializing a Ray Dataset. ``fn`` is expected to
  return the Dataset which is to be materialized. Runtime and throughput
  are automatically calculated and reported.
docstr_tok: ["Run","a","benchmark","on","materializing","a","Ray","Dataset",".","``","fn","``","is","expected","to","return","the","Dataset","which","is","to","be","materialized",".","Runtime","and","throughput","are","automatically","calculated","and","reported","."]
code: def run_materialize_ds(
      self,
      name: str,
      fn: Callable[..., Dataset],
      *fn_args,
      **fn_kwargs,
  ):
      """Run a benchmark on materializing a Ray Dataset. ``fn`` is expected to
      return the Dataset which is to be materialized. Runtime and throughput
      are automatically calculated and reported."""
  
      gc.collect()
  
      print(f"Running case: {name}")
      start_time = time.perf_counter()
      output_ds = fn(*fn_args, **fn_kwargs)
      output_ds.materialize()
      duration = time.perf_counter() - start_time
  
      # TODO(chengsu): Record more metrics based on dataset stats.
      num_rows = output_ds.count()
      self.result[name] = {
          BenchmarkMetric.RUNTIME.value: duration,
          BenchmarkMetric.NUM_ROWS.value: num_rows,
          BenchmarkMetric.THROUGHPUT.value: num_rows / duration,
      }
      print(f"Result of case {name}: {self.result[name]}")
  
file_path: ray/release/nightly_tests/dataset/benchmark.py
imports: import gc
  import json
  import os
  import time
  from typing import Callable, Dict
  from ray.data.dataset import Dataset
  from typing import Any
  from enum import Enum
  import pyarrow
  import pandas
label_desc: Use image node_id 2 for calling the Benchmark obj's underlying member method code with example usage: obj.run_materialize_ds(name, fn) without return types