code: def reconciliation(
      pdf: pd.DataFrame, hier_matrix: np.ndarray, target_col: str
  ):
      """
      Pyspark pandas UDF for computing the minimum-trace hierarchical time series reconciliation, as described by
      `Wickramasuriya et al. 2018 <https://robjhyndman.com/papers/mint.pdf>`__.
      Should be called on a pyspark dataframe grouped by timestamp. Pyspark implementation of
      `merlion.utils.hts.minT_reconciliation`.
  
      :param pdf: A ``pandas.DataFrame`` containing forecasted values & standard errors from ``m`` time series at a single
          timestamp. Each time series should be indexed by `TSID_COL_NAME`.
          The first ``n`` time series (in order of ID) orrespond to leaves of the hierarchy, while the remaining ``m - n``
          are weighted sums of the first ``n``.
          This dataframe can be produced by calling `forecast` on the dataframe produced by
          `merlion.spark.dataset.create_hier_dataset`.
      :param hier_matrix: A ``m``-by-``n`` matrix describing how the hierarchy is aggregated. The value of the ``k``-th
          time series is ``np.dot(hier_matrix[k], pdf[:n])``. This matrix can be produced by
          `merlion.spark.dataset.create_hier_dataset`.
      :param target_col: The name of the column whose value we wish to forecast.
  
      :return: A ``pandas.DataFrame`` which replaces the original forecasts & errors with reconciled forecasts & errors.
  
      .. note::
          Time series series reconciliation is skipped if the given timestamp has missing values for any of the
          time series. This can happen for training timestamps if the training time series has missing data and
          `forecast` is called with ``predict_on_train=true``.
      """
      # Get shape params & sort the data (for this timestamp) by time series ID.
      m, n = hier_matrix.shape
      assert len(pdf) <= m >= n
      if len(pdf) < m:
          return pdf
      assert (hier_matrix[:n] == np.eye(n)).all()
      pdf = pdf.sort_values(by=TSID_COL_NAME)
  
      # Compute the error weight matrix W (m by m)
      errname = f"{target_col}_err"
      coefs = hier_matrix.sum(axis=1)
      errs = (
          pdf[errname].values
          if errname in pdf.columns
          else np.full(m, np.nan)
      )
      nan_errs = np.isnan(errs)
      if nan_errs.all():
          W = np.diag(coefs)
      else:
          if nan_errs.any():
              errs[nan_errs] = (
                  np.nanmean(errs / coefs) * coefs[nan_errs]
              )
          W = np.diag(errs)
  
      # Create other supplementary matrices
      J = np.concatenate((np.eye(n), np.zeros((n, m - n))), axis=1)
      U = np.concatenate(
          (-hier_matrix[n:], np.eye(m - n)), axis=1
      )  # U.T from the paper
  
      # Compute projection matrix to compute coherent leaf forecasts
      inv = np.linalg.pinv(U @ W @ U.T)  # (m-n) by (m-n)
      P = J - ((J @ W) @ U.T) @ (inv @ U)  # n by m
  
      # Compute reconciled forecasts & errors
      rec = hier_matrix @ (P @ pdf[target_col].values)
      if nan_errs.all():
          rec_errs = errs
      else:
          # P * W.diagonal() is a faster way to compute P @ W, since W is diagonal
          rec_errs = hier_matrix @ (P * W.diagonal())  # m by m
          # np.sum(rec_errs ** 2, axis=1) is a faster way to compute (rec_errs @ rec_errs.T).diagonal()
          rec_errs = np.sqrt(np.sum(rec_errs**2, axis=1))
  
      # Replace original forecasts & errors with reconciled ones
      reconciled = pd.DataFrame(
          np.stack([rec, rec_errs], axis=1),
          index=pdf.index,
          columns=[target_col, errname],
      )
      df = pd.concat(
          (pdf.drop(columns=[target_col, errname]), reconciled), axis=1
      )
      return df
  
file_path: merlion/merlion/spark/pandas_udf.py
imports: import logging
  import traceback
  from typing import List, Union
  import numpy
  import pandas
  from merlion.models.factory import instantiate_or_copy_model, ModelFactory
  from merlion.models.anomaly.base import DetectorBase
  from merlion.models.forecast.base import ForecasterBase
  from merlion.spark.dataset import TSID_COL_NAME
  from merlion.utils import TimeSeries, to_pd_datetime
label_desc: Use image node_id 3 for calling a global function with example usage: reconciliation(pdf, hier_matrix, target_col) and returns: df, pdf