code: def calc_dp_rating_metrics(movie_views, backend, public_partitions):
      """Computes DP metrics."""
  
      # Set the total privacy budget.
      budget_accountant = pipeline_dp.NaiveBudgetAccountant(
          total_epsilon=1, total_delta=1e-6
      )
  
      # Create a DPEngine instance.
      dp_engine = pipeline_dp.DPEngine(budget_accountant, backend)
  
      # Specify which DP aggregated metrics to compute.
      params = pipeline_dp.AggregateParams(
          noise_kind=pipeline_dp.NoiseKind.LAPLACE,
          metrics=None,
          max_partitions_contributed=2,
          max_contributions_per_partition=1,
          min_value=1,
          max_value=5,
          public_partitions=public_partitions,
          custom_combiners=[CountCombiner()],
      )
  
      # Specify how to extract privacy_id, partition_key and value from an
      # element of movie view collection.
      data_extractors = pipeline_dp.DataExtractors(
          partition_extractor=lambda mv: mv.movie_id,
          privacy_id_extractor=lambda mv: mv.user_id,
          value_extractor=lambda mv: mv.rating,
      )
  
      # Run aggregation.
      dp_result = dp_engine.aggregate(
          movie_views, params, data_extractors
      )
  
      budget_accountant.compute_budgets()
  
      return dp_result
  
imports: from absl import app
  from absl import flags
  from apache_beam.runners.portability import fn_api_runner
  import pyspark
  from examples.movie_view_ratings.common_utils import *
  import pipeline_dp
  import numpy
label_desc: Use image node_id 2 for calling a global function with example usage: calc_dp_rating_metrics(movie_views, backend, public_partitions) and returns: dp_result