class_docstr: Class to observe (named) metrics on a :class:`DataFrame`.
  
  Metrics are aggregation expressions, which are applied to the DataFrame while it is being
  processed by an action.
  
  The metrics have the following guarantees:
  
  - It will compute the defined aggregates (metrics) on all the data that is flowing through
    the Dataset during the action.
  - It will report the value of the defined aggregate columns as soon as we reach the end of
    the action.
  
  The metrics columns must either contain a literal (e.g. lit(42)), or should contain one or
  more aggregate functions (e.g. sum(a) or sum(a + b) + avg(c) - lit(1)). Expressions that
  contain references to the input Dataset's columns must always be wrapped in an aggregate
  function.
  
  An Observation instance collects the metrics while the first action is executed. Subsequent
  actions do not modify the metrics returned by `Observation.get`. Retrieval of the metric via
  `Observation.get` blocks until the first action has finished and metrics become available.
  
  .. versionadded:: 3.3.0
  
  Notes
  -----
  This class does not support streaming datasets.
  
  Examples
  --------
  >>> from pyspark.sql.functions import col, count, lit, max
  >>> from pyspark.sql import Observation
  >>> df = spark.createDataFrame([["Alice", 2], ["Bob", 5]], ["name", "age"])
  >>> observation = Observation("my metrics")
  >>> observed_df = df.observe(observation, count(lit(1)).alias("count"), max(col("age")))
  >>> observed_df.count()
  2
  >>> observation.get
  {'count': 2, 'max(age)': 5}
docstr: Attaches this observation to the given :class:`DataFrame` to observe aggregations.
  
  Parameters
  ----------
  df : :class:`DataFrame`
      the :class:`DataFrame` to be observed
  exprs : list of :class:`Column`
      column expressions (:class:`Column`).
  
  Returns
  -------
  :class:`DataFrame`
      the observed :class:`DataFrame`.
docstr_tok: ["Attaches","this","observation","to","the","given",":","class",":","`","DataFrame","`","to","observe","aggregations",".","Parameters","--","--","--","--","--","df",":",":","class",":","`","DataFrame","`","the",":","class",":","`","DataFrame","`","to","be","observed","exprs",":","list","of",":","class",":","`","Column","`","column","expressions","(",":","class",":","`","Column","`",")",".","Returns","--","--","--","-",":","class",":","`","DataFrame","`","the","observed",":","class",":","`","DataFrame","`","."]
code: def _on(self, df: DataFrame, *exprs: Column) -> DataFrame:
      """Attaches this observation to the given :class:`DataFrame` to observe aggregations.
  
      Parameters
      ----------
      df : :class:`DataFrame`
          the :class:`DataFrame` to be observed
      exprs : list of :class:`Column`
          column expressions (:class:`Column`).
  
      Returns
      -------
      :class:`DataFrame`
          the observed :class:`DataFrame`.
      """
      if self._jo is not None:
          raise PySparkAssertionError(
              error_class="REUSE_OBSERVATION", message_parameters={}
          )
  
      self._jvm = df._sc._jvm
      assert self._jvm is not None
      cls = self._jvm.org.apache.spark.sql.Observation
      self._jo = cls(self._name) if self._name is not None else cls()
      observed_df = self._jo.on(
          df._jdf,
          exprs[0]._jc,
          column._to_seq(df._sc, [c._jc for c in exprs[1:]]),
      )
      return DataFrame(observed_df, df.sparkSession)
  
file_path: spark/python/pyspark/sql/observation.py
imports: import os
  from typing import Any, Dict, Optional
  from py4j.java_gateway import JavaObject, JVMView
  from pyspark.errors import PySparkTypeError, PySparkValueError, PySparkAssertionError
  from pyspark.sql import column
  from pyspark.sql.column import Column
  from pyspark.sql.dataframe import DataFrame
  from pyspark.sql.utils import is_remote
label_desc: Use image node_id 3 for calling the Observation obj's underlying member method code with example usage: obj._on(df) and returns: DataFrame