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 for a specific function; this is the most general
  benchmark utility available and will work if the other benchmark methods
  are too specific. However, ``fn`` is expected to return a
  `Dict[str, Any]` of metric labels to metric values, which are reported
  at the end of the benchmark. Runtime is automatically calculated and reported,
  but all other metrics of interest must be calculated and returned by ``fn``.
docstr_tok: ["Run","a","benchmark","for","a","specific","function",";","this","is","the","most","general","benchmark","utility","available","and","will","work","if","the","other","benchmark","methods","are","too","specific",".","However",",","``","fn","``","is","expected","to","return","a","`","Dict","[","str",",","Any","]","`","of","metric","labels","to","metric","values",",","which","are","reported","at","the","end","of","the","benchmark",".","Runtime","is","automatically","calculated","and","reported",",","but","all","other","metrics","of","interest","must","be","calculated","and","returned","by","``","fn","``","."]
code: def run_fn(
      self,
      name: str,
      fn: Callable[..., Dict[str, Any]],
      *fn_args,
      **fn_kwargs,
  ):
      """Run a benchmark for a specific function; this is the most general
      benchmark utility available and will work if the other benchmark methods
      are too specific. However, ``fn`` is expected to return a
      `Dict[str, Any]` of metric labels to metric values, which are reported
      at the end of the benchmark. Runtime is automatically calculated and reported,
      but all other metrics of interest must be calculated and returned by ``fn``.
      """
  
      gc.collect()
  
      print(f"Running case: {name}")
      start_time = time.perf_counter()
      # e.g. fn may output a dict of metrics
      fn_output = fn(*fn_args, **fn_kwargs)
      duration = time.perf_counter() - start_time
  
      curr_case_metrics = {
          BenchmarkMetric.RUNTIME.value: duration,
      }
      if isinstance(fn_output, dict):
          extra_metrics = {}
          for metric_key, metric_val in fn_output.items():
              if isinstance(metric_key, BenchmarkMetric):
                  curr_case_metrics[metric_key.value] = metric_val
              else:
                  extra_metrics[metric_key] = metric_val
          curr_case_metrics[
              BenchmarkMetric.EXTRA_METRICS.value
          ] = extra_metrics
  
      self.result[name] = curr_case_metrics
      print(f"Result of case {name}: {curr_case_metrics}")
  
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 4 for calling the Benchmark obj's underlying member method code with example usage: obj.run_fn(name, fn) without return types