class_docstr: DP Query on DataFrames.
  
  Warning: Don't create it directly, use QueryBuilder().
class_docstr_tok: ["DP","Query","on","DataFrames",".","Warning",":","Do","n't","create","it","directly",",","use","QueryBuilder","(",")","."]
docstr: Runs the DP query and returns a DataFrame.
  
  Args:
       budget: DP budget
       noise_kind: type of noise which is used for the anonymization.
  Returns:
      DataFrame with DP aggregation result.
docstr_tok: ["Runs","the","DP","query","and","returns","a","DataFrame",".","Args",":","budget",":","DP","budget","noise_kind",":","type","of","noise","which","is","used","for","the","anonymization",".","Returns",":","DataFrame","with","DP","aggregation","result","."]
code: def run_query(
      self,
      budget: Budget,
      noise_kind: pipeline_dp.NoiseKind = pipeline_dp.NoiseKind.LAPLACE,
  ):
      """Runs the DP query and returns a DataFrame.
  
      Args:
           budget: DP budget
           noise_kind: type of noise which is used for the anonymization.
      Returns:
          DataFrame with DP aggregation result.
      """
      converter = _create_dataframe_converter(self._df)
      backend = _create_backend_for_dataframe(self._df)
      col = converter.dataframe_to_collection(self._df, self._columns)
      budget_accountant = pipeline_dp.NaiveBudgetAccountant(
          total_epsilon=budget.epsilon, total_delta=budget.delta
      )
  
      dp_engine = pipeline_dp.DPEngine(budget_accountant, backend)
      metrics = list(self._metrics_output_columns.keys())
      params = pipeline_dp.AggregateParams(
          noise_kind=noise_kind,
          metrics=metrics,
          max_partitions_contributed=self._contribution_bounds.max_partitions_contributed,
          max_contributions_per_partition=self._contribution_bounds.max_contributions_per_partition,
          min_value=self._contribution_bounds.min_value,
          max_value=self._contribution_bounds.max_value,
      )
  
      data_extractors = pipeline_dp.DataExtractors(
          privacy_id_extractor=lambda row: row[0],
          partition_extractor=lambda row: row[1],
          value_extractor=lambda row: row[2],
      )
  
      dp_result = dp_engine.aggregate(
          col,
          params,
          data_extractors,
          public_partitions=self._public_partitions,
      )
      # dp_result: (partition_key, NamedTuple(metrics))
      budget_accountant.compute_budgets()
  
      # Convert elements to named tuple.
      metrics_names_to_output_columns = {}
      for metric, output_column in self._metrics_output_columns.items():
          metric_name = metric.name.lower()
          if output_column is None:
              output_column = metric_name  # default name
          metrics_names_to_output_columns[metric_name] = output_column
  
      output_columns = list(metrics_names_to_output_columns.values())
      partition_key = self._columns.partition_key
      partition_key_one_column = isinstance(partition_key, str)
      if partition_key_one_column:
          partition_key = [partition_key]
      PartitionMetricsTuple = namedtuple(
          "Result", partition_key + output_columns
      )
  
      def convert_to_partition_metrics_tuple(row):
          partition, metrics = row
          if partition_key_one_column:
              result = {partition_key[0]: partition}
          else:
              result = {}
              for key, value in zip(partition_key, partition):
                  result[key] = value
          for key, value in metrics._asdict().items():
              # Map default metric names to metric names specified in
              # self.metrics_names_to_output_columns
              result[metrics_names_to_output_columns[key]] = value
          return PartitionMetricsTuple(**result)
  
      dp_result = backend.map(
          dp_result,
          convert_to_partition_metrics_tuple,
          "Convert to NamedTuple",
      )
      # dp_result: PartitionMetricsTuple
  
      return converter.collection_to_dataframe(
          dp_result, output_columns
      )
  
imports: import abc
  import copy
  from collections import namedtuple
  from dataclasses import dataclass
  from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple, Union
  import pipeline_dp
  import pyspark
label_desc: Use image node_id 2 for calling the Query obj's underlying member method code with example usage: obj.run_query(budget, noise_kind) and returns: converter, PartitionMetricsTuple