code: def anomaly(
      pdf: pd.DataFrame,
      index_cols: List[str],
      time_col: str,
      train_test_split: Union[int, str],
      model: Union[DetectorBase, dict],
      predict_on_train: bool = False,
  ) -> pd.DataFrame:
      """
      Pyspark pandas UDF for performing anomaly detection.
      Should be called on a pyspark dataframe grouped by time series ID, i.e. by ``index_cols``.
  
      :param pdf: The ``pandas.DataFrame`` containing the training and testing data. Should be a single time series.
      :param index_cols: The list of column names used to index all the time series in the dataset. Not used for modeling.
      :param time_col: The name of the column containing the timestamps.
      :param train_test_split: The time at which the testing data starts.
      :param model: The model (or model ``dict``) we are using to predict anomaly scores.
      :param predict_on_train: Whether to return the model's prediction on the training data.
  
      :return: A ``pandas.DataFrame`` with the anomaly scores on the test data.
          Columns are ``[*index_cols, time_col, \"anom_score\"]``.
      """
      # Sort the dataframe by time & turn it into a Merlion time series
      if (
          TSID_COL_NAME not in index_cols
          and TSID_COL_NAME in pdf.columns
      ):
          index_cols = index_cols + [TSID_COL_NAME]
      pdf = pdf.sort_values(by=time_col)
      ts = TimeSeries.from_pd(
          pdf.drop(columns=index_cols).set_index(time_col)
      )
  
      # Create model
      model = instantiate_or_copy_model(
          model or {"name": "DefaultDetector"}
      )
      if not isinstance(model, DetectorBase):
          raise TypeError(
              f"Expected `model` to be an instance of DetectorBase, but got {model}."
          )
  
      # Train model & run inference
      exception = False
      train, test = ts.bisect(train_test_split, t_in_left=False)
      try:
          train_pred = model.train(train)
          train_pred = model.post_rule(train_pred).to_pd()
          pred = model.get_anomaly_label(test).to_pd()
      except Exception:
          exception = True
          row0 = pdf.iloc[0]
          idx = ", ".join(f"{k} = {row0[k]}" for k in index_cols)
          logger.warning(
              f"Model {type(model).__name__} threw an exception on ({idx}). {traceback.format_exc()}"
              f"Trying StatThreshold model instead.\n"
          )
      if exception:
          try:
              model = ModelFactory.create(
                  name="StatThreshold",
                  target_seq_index=0,
                  threshold=model.threshold,
              )
              train_pred = model.train(train)
              train_pred = model.post_rule(train_pred).to_pd()
              pred = model.get_anomaly_label(test).to_pd()
          except Exception:
              logger.warning(
                  f"Model StatThreshold threw an exception on ({idx}).{traceback.format_exc()}"
                  f"Returning anomaly score = 0 as a placeholder.\n"
              )
              train_pred = pd.DataFrame(
                  0,
                  index=to_pd_datetime(train.time_stamps),
                  columns=["anom_score"],
              )
              pred = pd.DataFrame(
                  0,
                  index=to_pd_datetime(test.time_stamps),
                  columns=["anom_score"],
              )
  
      if predict_on_train and train_pred is not None:
          pred = pd.concat((train_pred, pred))
  
      # Turn the time index into a regular column, and add the index columns back to the prediction
      pred.index.name = time_col
      pred.reset_index(inplace=True)
      index_pdf = pd.concat(
          [pdf[index_cols].iloc[:1]] * len(pred), ignore_index=True
      )
      return pd.concat((index_pdf, pred), axis=1)
  
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 2 for calling a global function with example usage: anomaly(pdf, index_cols, time_col, train_test_split, model, predict_on_train) and returns: pd