class_docstr: A helper class that contains methods to for privacy utility analysis.
docstr: Generates sketches in the format of (partition_key, value, partition_count).
  
  The sketches has one entry for each unique (partition_key, privacy_id).
  Parameter tuning on outputs of sketch ignores `min_value` and `max_value` of
  AggregateParams
  
  partition_key: the hashed version of the current partition key
  partition_value: the per privacy id per partition_key aggregated value
  partition_count: the number of partitions this privacy id contributes to
  
  Args:
    input_data: The data to sample. It can be local data, beam PCollection or
      Spark RDD depending on the engine used.
    params: The parameters defining sampling properties.
    data_extractors: A function to extract privacy_id, partition_key, value of
      the input data.
  
  Returns:
    Sketches in the format of (partition_key, value, partition_count).
code: def sketch(
      self,
      input_data: DataType,
      params: SampleParams,
      data_extractors: pipeline_dp.DataExtractors,
  ) -> DataType:
      """Generates sketches in the format of (partition_key, value, partition_count).
  
      The sketches has one entry for each unique (partition_key, privacy_id).
      Parameter tuning on outputs of sketch ignores `min_value` and `max_value` of
      AggregateParams
  
      partition_key: the hashed version of the current partition key
      partition_value: the per privacy id per partition_key aggregated value
      partition_count: the number of partitions this privacy id contributes to
  
      Args:
        input_data: The data to sample. It can be local data, beam PCollection or
          Spark RDD depending on the engine used.
        params: The parameters defining sampling properties.
        data_extractors: A function to extract privacy_id, partition_key, value of
          the input data.
  
      Returns:
        Sketches in the format of (partition_key, value, partition_count).
      """
      if params.metrics is None:
          raise ValueError(
              "Must provide aggregation metrics for sketch."
          )
      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."
          )
      combiner = non_private_combiners.create_compound_combiner(
          metrics=params.metrics
      )
  
      # Extract the columns.
      col = self._be.map(
          input_data,
          functools.partial(_extract_fn, data_extractors),
          "Extract (privacy_id, partition_key, value))",
      )
      # col : (privacy_id, partition_key, value)
      col = self._be.map_tuple(
          col,
          lambda pid, pk, v: (pk, (pid, v)),
          "Rekey to (partition_key, (privacy_id, value))",
      )
      # col : (partition_key, (privacy_id, value))
      # sample
      # group by key, filter keys by sampling, expand the values by flat map
      col = self._be.group_by_key(col, "Group by pk")
      col = self._be.map_tuple(
          col,
          lambda pk, pid_v_seq: (1, (pk, pid_v_seq)),
          "Rekey to (1, (pk, pid_v_seq))",
      )
      col = self._be.sample_fixed_per_key(
          col, params.number_of_sampled_partitions, "Sample partitions"
      )
      col = self._be.flat_map(
          col, lambda plst: plst[1], "Extract values"
      )
  
      def flatten_sampled_results(
          pk_pid_pval_list: Tuple[Any, Sequence[Tuple[Any, Any]]]
      ) -> Sequence[Tuple[Any, Tuple[Any, Any]]]:
          pk, pid_pval_list = pk_pid_pval_list
          return [(pk, pid_pval) for pid_pval in pid_pval_list]
  
      col = self._be.flat_map(
          col, flatten_sampled_results, "Flatten to (pk, (pid, value))"
      )
  
      # col : (partition_key, (privacy_id, value))
      # calculates partition_count after sampling and per
      # (partition_key, privacy_id) pair aggregated value
      col = self._be.map_tuple(
          col,
          lambda pk, pid_v: ((pk, pid_v[0]), pid_v[1]),
          "Transform to (pk, pid), value))",
      )
      # col : ((partition_key, privacy_id), value)
  
      col = self._be.group_by_key(col, "Group by (pk, pid)")
      # col : ((partition_key, privacy_id), [value])
      col = self._be.map_values(
          col, combiner.create_accumulator, "Aggregate by (pk, pid)"
      )
      # col : ((partition_key, privacy_id), accumulator)
      col = self._be.map_tuple(
          col,
          lambda pk_pid, p_value: (pk_pid[1], (pk_pid[0], p_value)),
          "Transform to (pid, (pk, accumulator))",
      )
      # col : (privacy_id, (partition_key, accumulator))
      col = self._be.group_by_key(col, "Group by privacy_id")
  
      key_accumulator_sequence_type = Sequence[
          Tuple[Any, pipeline_dp.accumulator.Accumulator]
      ]
  
      def calculate_partition_count(
          key_accumulator_list: key_accumulator_sequence_type,
      ) -> Tuple[int, key_accumulator_sequence_type]:
          partition_count = len(
              set(pk for pk, _ in key_accumulator_list)
          )
          return (partition_count, key_accumulator_list)
  
      col = self._be.map_values(
          col, calculate_partition_count, "Calculates partition_count"
      )
  
      # col : (privacy_id, (partition_count, [(partition_key, accumulator)]))
  
      def flatten_results(
          input_col: Tuple[
              Any, Tuple[int, key_accumulator_sequence_type]
          ]
      ) -> Sequence[Tuple[Any, Any, int]]:
          _, pcount_pk_acc_list = input_col
          pcount, pk_acc_list = pcount_pk_acc_list
          return [(pk, acc[0], pcount) for pk, acc in pk_acc_list]
  
      return self._be.flat_map(
          col,
          flatten_results,
          "Flatten to (pk, aggregated_value, partition_count)",
      )
  
file_path: pipelinedp/utility_analysis/data_peeker.py
imports: import dataclasses
  import functools
  from typing import Any, Sequence, Union, Optional, Tuple
  import pipeline_dp
  from utility_analysis import non_private_combiners
label_desc: Use image node_id 2 for calling the DataPeeker obj's underlying member method code with example usage: obj.sketch(input_data, params, data_extractors) and returns: self, unknown, partition_count,  key_accumulator_list, unknown