code: def merge_map_results(
      self,
      dataset,
      metric_names,
      metric_types,
      save_path,
      num_workers,
      num_threads,
      num_threads_reduce,
  ):
      total_num_samples = len(dataset)
      sample_idx_dtype = find_fit_int_dtype(0, total_num_samples - 1)
      logger.info(
          f"Total number of data samples: {total_num_samples}. Will use {sample_idx_dtype} to store the sample indexes."
      )
      for m_idx in range(len(metric_names)):
          metric_name, metric_type = (
              metric_names[m_idx],
              metric_types[m_idx],
          )
          if metric_type == "single_value_per_sample":
              metric_save_path = f"{save_path}/{metric_name}/"
              sample_to_metric_count = 0
              unique_metric_values = set([])
              manager = Manager()
              return_dict = manager.dict()
              p = []
              for t_idx_reduce in range(num_threads_reduce):
                  p.append(
                      Process(
                          target=self.merge_gather_map_stats,
                          args=(
                              num_workers,
                              num_threads,
                              num_threads_reduce,
                              t_idx_reduce,
                              metric_save_path,
                              metric_name,
                              return_dict,
                          ),
                      )
                  )
                  p[t_idx_reduce].start()
              for t_idx_reduce in range(num_threads_reduce):
                  p[t_idx_reduce].join()
              for t_idx_reduce in range(num_threads_reduce):
                  results = return_dict[t_idx_reduce]
                  for res in results:
                      unique_metric_values = unique_metric_values.union(
                          set(res[0])
                      )
                      sample_to_metric_count += res[1]
              value_max = max(unique_metric_values)
              value_min = min(unique_metric_values)
              assert (
                  sample_to_metric_count == total_num_samples
              ), "The number of samples in map result files are not correct. It's possible that some map worker didn't finish successfully."
              metric_value_dtype = find_fit_int_dtype(
                  value_min, value_max
              )
              logger.info(
                  f"Metric {metric_name} has values between {value_min} and {value_max}. Will use {metric_value_dtype} to store the metric values."
              )
  
              # sample_to_metric
              map_worker_thread = []
              for w_idx in range(num_workers):
                  for t_idx in range(num_threads):
                      map_worker_thread.append([w_idx, t_idx])
              thread_splits = split_index(
                  0, len(map_worker_thread), num_threads_reduce
              )
              p = []
              for t_idx_reduce in range(num_threads_reduce):
                  start_idx, end_idx = (
                      thread_splits[t_idx_reduce][0],
                      thread_splits[t_idx_reduce][1],
                  )
                  p.append(
                      Process(
                          target=self.merge_sample_to_metric,
                          args=(
                              t_idx_reduce,
                              metric_save_path,
                              metric_name,
                              metric_value_dtype,
                              map_worker_thread[start_idx:end_idx],
                          ),
                      )
                  )
                  p[t_idx_reduce].start()
              for t_idx_reduce in range(num_threads_reduce):
                  p[t_idx_reduce].join()
  
              sample_to_metric_fname = (
                  f"{metric_save_path}/{metric_name}_sample_to_metric"
              )
              sample_to_metric_builder = create_mmap_dataset_builder(
                  sample_to_metric_fname, metric_value_dtype
              )
              for t_idx_reduce in range(num_threads_reduce):
                  chunk_fname = f"{metric_save_path}/{metric_name}_sample_to_metric_thread{t_idx_reduce}"
                  logger.info(f"Merging file {chunk_fname}")
                  sample_to_metric_builder.merge_file_(chunk_fname)
              close_mmap_dataset_builder(
                  sample_to_metric_builder, sample_to_metric_fname
              )
              sample_to_metric = MMapIndexedDataset(
                  sample_to_metric_fname, skip_warmup=True
              )
              assert len(sample_to_metric) == total_num_samples
  
              # metric_to_sample
              unique_metric_values = list(sorted(unique_metric_values))
              thread_splits = split_index(
                  0, len(unique_metric_values), num_threads_reduce
              )
              p = []
              for t_idx_reduce in range(num_threads_reduce):
                  start_idx, end_idx = (
                      thread_splits[t_idx_reduce][0],
                      thread_splits[t_idx_reduce][1],
                  )
                  p.append(
                      Process(
                          target=self.merge_metric_to_sample,
                          args=(
                              t_idx_reduce,
                              metric_save_path,
                              metric_name,
                              sample_idx_dtype,
                              metric_value_dtype,
                              unique_metric_values[start_idx:end_idx],
                              num_workers,
                              num_threads,
                          ),
                      )
                  )
                  p[t_idx_reduce].start()
              for t_idx_reduce in range(num_threads_reduce):
                  p[t_idx_reduce].join()
              index_to_sample_fname = (
                  f"{metric_save_path}/{metric_name}_index_to_sample"
              )
              index_to_sample_builder = create_mmap_dataset_builder(
                  index_to_sample_fname, sample_idx_dtype
              )
              index_to_metric_fname = (
                  f"{metric_save_path}/{metric_name}_index_to_metric"
              )
              index_to_metric_builder = create_mmap_dataset_builder(
                  index_to_metric_fname, metric_value_dtype
              )
              for t_idx_reduce in range(num_threads_reduce):
                  chunk_is_fname = f"{metric_save_path}/{metric_name}_index_to_sample_thread{t_idx_reduce}"
                  logger.info(f"Merging file {chunk_is_fname}")
                  index_to_sample_builder.merge_file_(chunk_is_fname)
                  chunk_im_fname = f"{metric_save_path}/{metric_name}_index_to_metric_thread{t_idx_reduce}"
                  logger.info(f"Merging file {chunk_im_fname}")
                  index_to_metric_builder.merge_file_(chunk_im_fname)
              close_mmap_dataset_builder(
                  index_to_sample_builder, index_to_sample_fname
              )
              close_mmap_dataset_builder(
                  index_to_metric_builder, index_to_metric_fname
              )
              num_sample_per_value = {}
              index_to_sample = MMapIndexedDataset(
                  index_to_sample_fname, skip_warmup=True
              )
              index_to_metric = MMapIndexedDataset(
                  index_to_metric_fname, skip_warmup=True
              )
              index_to_sample_merged_fname = f"{metric_save_path}/{metric_name}_index_to_sample_percentile_merged"
              index_to_sample_merged_builder = (
                  create_mmap_dataset_builder(
                      index_to_sample_merged_fname, sample_idx_dtype
                  )
              )
              for v_idx in range(len(index_to_sample)):
                  if v_idx > 0:
                      assert (
                          index_to_metric[v_idx]
                          > index_to_metric[v_idx - 1]
                      )
                  num_sample_per_value[index_to_metric[v_idx][0]] = len(
                      index_to_sample[v_idx]
                  )
              assert (
                  sum(num_sample_per_value.values())
                  == total_num_samples
              )
              merge_step = max(1, len(index_to_sample) // 100)
              for v_idx in range(0, len(index_to_sample), merge_step):
                  merged_samples = np.copy(
                      np.concatenate(
                          index_to_sample[
                              v_idx : min(
                                  len(index_to_sample),
                                  (v_idx + merge_step),
                              )
                          ],
                          axis=None,
                      )
                  )
                  index_to_sample_merged_builder.add_item(
                      torch.tensor(
                          merged_samples.astype(np.int64),
                          dtype=torch.long,
                      )
                  )
                  logger.info(
                      f"Finished merging index_to_sample {v_idx} to {v_idx+merge_step}."
                  )
              close_mmap_dataset_builder(
                  index_to_sample_merged_builder,
                  index_to_sample_merged_fname,
              )
              self.get_metric_value_percentiles(
                  metric_name, num_sample_per_value, total_num_samples
              )
          elif metric_type == "accumulate_value_over_samples":
              metric_save_path = f"{save_path}/{metric_name}/"
              metric_value = None
              for w_idx in range(num_workers):
                  for t_idx in range(num_threads):
                      w_metric_save_path = f"{metric_save_path}/worker{w_idx}_thread{t_idx}/"
                      w_metric_value_fname = f"{w_metric_save_path}/{metric_name}_metric_value"
                      w_metric_value = MMapIndexedDataset(
                          w_metric_value_fname, skip_warmup=True
                      )
                      if metric_value is None:
                          metric_value = np.copy(w_metric_value[0])
                      else:
                          metric_value += np.copy(w_metric_value[0])
              value_max = int(max(metric_value))
              value_min = int(min(metric_value))
              metric_value_dtype = find_fit_int_dtype(
                  value_min, value_max
              )
              metric_value_fname = (
                  f"{metric_save_path}/{metric_name}_metric_value"
              )
              metric_value_builder = create_mmap_dataset_builder(
                  metric_value_fname, metric_value_dtype
              )
              metric_value_builder.add_item(
                  torch.tensor(
                      metric_value.astype(np.int64), dtype=torch.long
                  )
              )
              close_mmap_dataset_builder(
                  metric_value_builder, metric_value_fname
              )
  
imports: import os
  from collections import defaultdict
  import csv
  import time
  from multiprocessing import Process, Manager
  import numpy
  import torch
  from torch.utils.data import BatchSampler, SequentialSampler, DataLoader, Subset
  from deepspeed.utils import logger
  from .indexed_dataset import MMapIndexedDataset
  from .utils import split_dataset, split_index, create_mmap_dataset_builder, close_mmap_dataset_builder, find_fit_int_dtype
label_desc: Use image node_id 11 for calling the DataAnalyzer obj's underlying member method code with example usage: obj.merge_map_results(dataset, metric_names, metric_types, save_path, num_workers, num_threads, num_threads_reduce) without return types