class_docstr: Performs aggregations for utility analysis.
class_docstr_tok: ["Performs","aggregations","for","utility","analysis","."]
docstr: Computes approximated DP aggregation on sketches generated by peeker.
  
  The input `col` has one entry for each unique (partition_key, privacy_id).
  They are in the format of (partition_key, partition_value, partition_count).
  
  partition_key: the hashed version of the current partition key
  partition_value: the per privacy id per partition_key aggregated value
  partition_count: number of partitions this privacy id contributes to
  
  Note, this is not a DP aggregation. There are certain shortcuts made
  to make sure utility analysis run fast in realtime. The output is intended
  for the utility analysis. Please don't release metrics generated by this
  funciton if you want DP guarantee.
  
  Note, this function only supports one aggregation that needs to be either
  count or sum. This is determined by the nature of sketching.
  
  Args:
      col: Sketches in the format of (partition_key, partition_value,
          partition_count).
      params: Specifies which metrics to compute and computation parameters.
  
  Returns:
      Aggregation metrics.
docstr_tok: ["Computes","approximated","DP","aggregation","on","sketches","generated","by","peeker",".","The","input","`","col","`","has","one","entry","for","each","unique","(","partition_key",",","privacy_id",")",".","They","are","in","the","format","of","(","partition_key",",","partition_value",",","partition_count",")",".","partition_key",":","the","hashed","version","of","the","current","partition","key","partition_value",":","the","per","privacy","id","per","partition_key","aggregated","value","partition_count",":","number","of","partitions","this","privacy","id","contributes","to","Note",",","this","is","not","a","DP","aggregation",".","There","are","certain","shortcuts","made","to","make","sure","utility","analysis","run","fast","in","realtime",".","The","output","is","intended","for","the","utility","analysis",".","Please","do","n't","release","metrics","generated","by","this","funciton","if","you","want","DP","guarantee",".","Note",",","this","function","only","supports","one","aggregation","that","needs","to","be","either","count","or","sum",".","This","is","determined","by","the","nature","of","sketching",".","Args",":","col",":","Sketches","in","the","format","of","(","partition_key",",","partition_value",",","partition_count",")",".","params",":","Specifies","which","metrics","to","compute","and","computation","parameters",".","Returns",":","Aggregation","metrics","."]
code: def aggregate_sketches(
      self, col, params: pipeline_dp.AggregateParams
  ) -> Sequence[Tuple[Any, Sequence[int]]]:
      """Computes approximated DP aggregation on sketches generated by peeker.
  
      The input `col` has one entry for each unique (partition_key, privacy_id).
      They are in the format of (partition_key, partition_value, partition_count).
  
      partition_key: the hashed version of the current partition key
      partition_value: the per privacy id per partition_key aggregated value
      partition_count: number of partitions this privacy id contributes to
  
      Note, this is not a DP aggregation. There are certain shortcuts made
      to make sure utility analysis run fast in realtime. The output is intended
      for the utility analysis. Please don't release metrics generated by this
      funciton if you want DP guarantee.
  
      Note, this function only supports one aggregation that needs to be either
      count or sum. This is determined by the nature of sketching.
  
      Args:
          col: Sketches in the format of (partition_key, partition_value,
              partition_count).
          params: Specifies which metrics to compute and computation parameters.
  
      Returns:
          Aggregation metrics.
      """
      if len(params.metrics) != 1 or params.metrics[0] not in [
          pipeline_dp.aggregate_params.Metrics.SUM,
          pipeline_dp.aggregate_params.Metrics.COUNT,
      ]:
          raise ValueError(
              "Sketch only supports a single aggregation and it must be COUNT or SUM."
          )
  
      # col: (partition_key, per_user_aggregated_value, partition_count)
      combiner = pipeline_dp.combiners.create_compound_combiner(
          params, self._budget_accountant
      )
  
      col = self._ops.filter(
          col,
          functools.partial(
              _cross_partition_filter_fn,
              params.max_partitions_contributed,
          ),
          "Cross partition bounding",
      )
      # TODO: use max_contributions_per_partition for COUNT and max_sum_per_partition for SUM.
      col = self._ops.map_tuple(
          col,
          functools.partial(
              _per_partition_bounding,
              params.max_contributions_per_partition,
          ),
          "Per partition bounding",
      )
      # col: (partition_key, per_user_aggregated_value)
      col = self._ops.map_values(
          col,
          lambda x: (1, (x,)),
          "Convert to format of CompoundCombiner",
      )
      # col: (partition_key, compound_accumulator)
      col = self._ops.combine_accumulators_per_key(
          col, combiner, "Aggregate by partition key"
      )
      # col: (partition_key, aggregated_compound_accumulator)
      # Partition selection
      budget = self._budget_accountant.request_budget(
          mechanism_type=pipeline_dp.MechanismType.GENERIC
      )
  
      partition_selection_filter_fn = functools.partial(
          _partition_selection_filter_fn,
          budget,
          params.max_partitions_contributed,
      )
  
      col = self._ops.filter(
          col,
          partition_selection_filter_fn,
          "Filter private partitions",
      )
      # col: (partition_key, aggregated_compound_accumulator)
      col = self._ops.map_values(
          col, combiner.compute_metrics, "Compute DP metrics"
      )
      # col: (partition_key, metrics)
      return col
  
file_path: pipelinedp/utility_analysis/peeker_engine.py
imports: import functools
  from typing import Any, Sequence, Tuple, Union
  from apache_beam import pvalue
  import numpy
  import pydp
  import pipeline_dp
label_desc: Use image node_id 2 for calling the PeekerEngine obj's underlying member method code with example usage: obj.aggregate_sketches(col, params) and returns: col