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"
              >> MakePrivate(
                  budget_accountant=budget_accountant,
                  privacy_id_extractor=lambda mv: mv.user_id,
              )
          )
  
          explain_computation_report = (
              pipeline_dp.ExplainComputationReport()
          )
          # Calculate the private sum
          dp_result = private_movie_views | "Private Sum" >> private_beam.Sum(
              SumParams(
                  # 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,
                  # .. with minimal rating of "1"
                  min_value=1,
                  # .. and maximum rating of "5"
                  max_value=5,
                  # The aggregation key: we're grouping data by movies
                  partition_extractor=lambda mv: mv.movie_id,
                  # The value we're aggregating: we're summing up ratings
                  value_extractor=lambda mv: mv.rating,
              )
          )
          budget_accountant.compute_budgets()
  
          # Generate the Explain Computation Report. It must be called after
          # budget_accountant.compute_budgets().
          print(explain_computation_report.text())
  
          # Save the results
          dp_result | beam.io.WriteToText(FLAGS.output_file)
  
      return 0
  
file_path: pipelinedp/examples/movie_view_ratings/run_on_beam.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 pipeline_dp import SumParams
  from pipeline_dp.private_beam import MakePrivate
  from common_utils import ParseFile
label_desc: Use image node_id 1 for calling a global function with example usage: main(unused_argv) and returns: int