code: def main(unused_argv):
      # Setup Beam
  
      # Here, we use a local Beam runner.
      # For a truly distributed calculation, connect to a Beam cluster (e.g.
      # running on some cloud provider).
      runner = fn_api_runner.FnApiRunner()  # Local Beam runner
      with beam.Pipeline(runner=runner) as pipeline:
          # Define the privacy budget available for our computation.
          budget_accountant = pipeline_dp.NaiveBudgetAccountant(
              total_epsilon=1, total_delta=1e-6
          )
  
          # Load and parse input data
          movie_views_pcol = (
              pipeline
              | beam.io.ReadFromText(FLAGS.input_file)
              | beam.ParDo(ParseFile())
          )
  
          # Wrap Beam's PCollection into it's private version
          private_movie_views = (
              movie_views_pcol
              | "Create private collection"
              >> pbeam.MakePrivate(
                  budget_accountant=budget_accountant,
                  privacy_id_extractor=lambda mv: mv.user_id,
              )
          )
  
          private_movie_views = private_movie_views | pbeam.Map(
              lambda mv: (mv.movie_id, mv.rating)
          )
  
          # Calculate the private sum
          dp_result = private_movie_views | pbeam.CombinePerKey(
              DPSumCombineFn(min_value=1, max_value=5),
              pbeam.CombinePerKeyParams(
                  # Limits to how much one user can contribute:
                  # .. at most two movies rated per user
                  max_partitions_contributed=2,
                  # .. at most one rating for each movie
                  max_contributions_per_partition=1,
              ),
          )
          budget_accountant.compute_budgets()
  
          # Save the results
          dp_result | beam.io.WriteToText(FLAGS.output_file)
  
      return 0
  
file_path: pipelinedp/examples/experimental/beam_combine_fn.py
imports: from absl import app
  from absl import flags
  import apache_beam
  from apache_beam.runners.portability import fn_api_runner
  import pipeline_dp
  from pipeline_dp import private_beam
  from examples.movie_view_ratings.common_utils import ParseFile
  import numpy
  from pydp.algorithms import numerical_mechanisms
label_desc: Use image node_id 1 for calling a global function with example usage: main(unused_argv) and returns: int