class_docstr: Bounds the contribution of privacy_id cross partitions.
  
  It ensures that each privacy_id contributes to not more than
  max_partitions_contributed partitions (cross-partition contribution
  bounding), by performing sampling if needed. It is assumed that the provided
  aggregate_fn function bounds per-partition contributions.
class_docstr_tok: ["Bounds","the","contribution","of","privacy_id","cross","partitions",".","It","ensures","that","each","privacy_id","contributes","to","not","more","than","max_partitions_contributed","partitions","(","cross-partition","contribution","bounding",")",",","by","performing","sampling","if","needed",".","It","is","assumed","that","the","provided","aggregate_fn","function","bounds","per-partition","contributions","."]
code: def bound_contributions(
      self, col, params, backend, report_generator, aggregate_fn
  ):
      col = backend.map_tuple(
          col,
          lambda pid, pk, v: (pid, (pk, v)),
          "Rekey to ((privacy_id), (partition_key, value))",
      )
  
      col = backend.group_by_key(col, "Group by privacy_id")
      # (privacy_id, [(partition_key, value)])
  
      col = collect_values_per_partition_key_per_privacy_id(
          col, backend
      )
      # (privacy_id, [(partition_key, [value])])
  
      # Bound cross partition contributions with sampling.
      sample = sampling_utils.choose_from_list_without_replacement
      sample_size = params.max_partitions_contributed
      col = backend.map_values(
          col, lambda a: sample(a, sample_size), "Sample"
      )
  
      # (privacy_id, [partition_key, [value]])
  
      # Unnest the list per privacy id.
      def rekey_per_privacy_id_per_partition_key(pid_pk_v_values):
          privacy_id, partition_values = pid_pk_v_values
  
          for partition_key, values in partition_values:
              yield (privacy_id, partition_key), values
  
      col = backend.flat_map(
          col,
          rekey_per_privacy_id_per_partition_key,
          "Unnest per privacy_id",
      )
      # ((privacy_id, partition_key), [value])
  
      return backend.map_values(
          col,
          aggregate_fn,
          "Apply aggregate_fn after cross-partition contribution bounding",
      )
  
file_path: pipelinedp/pipeline_dp/contribution_bounders.py
imports: import abc
  import collections
  from typing import Callable, Iterable
  import pipeline_dp
  from pipeline_dp import pipeline_backend
  from pipeline_dp import sampling_utils
label_desc: Use image node_id 1 for calling the SamplingCrossPartitionContributionBounder obj's underlying member method code with example usage: obj.bound_contributions(col, params, backend, report_generator, aggregate_fn) and returns: backend