code: def perform_utility_analysis(
      col,
      backend: pipeline_backend.PipelineBackend,
      options: analysis.UtilityAnalysisOptions,
      data_extractors: Union[
          pipeline_dp.DataExtractors, pipeline_dp.PreAggregateExtractors
      ],
      public_partitions=None,
  ):
      """Performs utility analysis for DP aggregations.
  
      Args:
        col: collection where all elements are of the same type.
        backend: PipelineBackend with which the utility analysis will be run.
        options: options for utility analysis.
        data_extractors: functions that extract needed pieces of information
          from elements of 'col'. In case if the analysis performed on
          pre-aggregated data, it should have type PreAggregateExtractors
          otherwise DataExtractors.
        public_partitions: A collection of partition keys that will be present
          in the result. If not provided, the utility analysis with private
          partition selection will be performed.
      Returns:
          returns a tuple. Its 1st element is a collection which contains
            metrics.UtilityReport with one report per each input configuration.
            The 2nd element of the tuple is a collection with elements
            ((partition_key, configuration_index), metrics.PerPartitionMetrics).
      """
      budget_accountant = pipeline_dp.NaiveBudgetAccountant(
          total_epsilon=options.epsilon, total_delta=options.delta
      )
      engine = utility_analysis_engine.UtilityAnalysisEngine(
          budget_accountant=budget_accountant, backend=backend
      )
      per_partition_result = engine.analyze(
          col,
          options=options,
          data_extractors=data_extractors,
          public_partitions=public_partitions,
      )
      # (partition_key, utility results)
      budget_accountant.compute_budgets()
  
      n_configurations = options.n_configurations
      per_partition_result = backend.map_values(
          per_partition_result,
          lambda value: _pack_per_partition_metrics(
              value, n_configurations
          ),
          "Pack per-partition metrics.",
      )
      # (partition_key, (metrics.PerPartitionMetrics))
  
      per_partition_result = backend.to_multi_transformable_collection(
          per_partition_result
      )
      # (partition_key, (metrics.PerPartitionMetrics))
  
      col = backend.values(per_partition_result, "Drop partition key")
      # ((metrics.PerPartitionMetrics))
  
      col = backend.flat_map(col, _unnest_metrics, "Unnest metrics")
      # ((configuration_index, bucket), metrics.PerPartitionMetrics)
  
      per_partition_result = backend.flat_map(
          per_partition_result,
          lambda kv: (
              ((kv[0], i), result) for i, result in enumerate(kv[1])
          ),
          "Unpack PerPartitionMetrics from list",
      )
      # ((partition_key, configuration_index), metrics.PerPartitionMetrics)
  
      combiner = cross_partition_combiners.CrossPartitionCombiner(
          options.aggregate_params.metrics,
          public_partitions is not None,
      )
  
      accumulators = backend.map_values(
          col, combiner.create_accumulator, "Create accumulators"
      )
      # accumulators : ((configuration_index, bucket), accumulator)
  
      accumulators = backend.combine_accumulators_per_key(
          accumulators, combiner, "Combine cross-partition metrics"
      )
      # accumulators : ((configuration_index, bucket), accumulator)
  
      cross_partition_metrics = backend.map_values(
          accumulators,
          combiner.compute_metrics,
          "Compute cross-partition metrics",
      )
      #  ((configuration_index, bucket), UtilityReport)
  
      cross_partition_metrics = backend.map_tuple(
          cross_partition_metrics,
          lambda key, value: (key[0], (key[1], value)),
          "Rekey",
      )
      # (configuration_index, (bucket, UtilityReport))
  
      cross_partition_metrics = backend.group_by_key(
          cross_partition_metrics, "Group by configuration"
      )
      # (configuration_index, Iterable[(bucket, UtilityReport)])
      result = backend.map_tuple(
          cross_partition_metrics,
          _group_utility_reports,
          "Group utility reports",
      )
      if public_partitions is None:
          # Add partition selection strategy for private partitions.
          strategies = data_structures.get_partition_selection_strategy(
              options
          )
  
          def add_partition_selection_strategy(
              report: metrics.UtilityReport,
          ):
              # Beam does not allow to change input arguments in map, so copy it.
              report = copy.deepcopy(report)
              strategy = strategies[report.configuration_index]
              report.partitions_info.strategy = strategy
              for bin in report.utility_report_histogram:
                  bin.report.partitions_info.strategy = strategy
              return report
  
          result = backend.map(
              result,
              add_partition_selection_strategy,
              "Add Partition Selection Strategy",
          )
          # result: (UtilityReport)
      # result: (UtilityReport)
      return result, per_partition_result
  
file_path: pipelinedp/analysis/utility_analysis.py
imports: from typing import Any, Iterable, List, Tuple, Union
  import pipeline_dp
  from pipeline_dp import pipeline_backend
  import analysis
  from analysis import data_structures
  from analysis import metrics
  from analysis import utility_analysis_engine
  from analysis import cross_partition_combiners
  import copy
  import bisect
label_desc: Use image node_id 2 for calling a global function with example usage: perform_utility_analysis(col, backend, options, data_extractors, public_partitions) and returns: result,  per_partition_result, report