Hey community members

I am new to Databricks and was building a simple DLT pipleine that loads data from S3 and runs an Isolation forest prediction to detect anomalies. The model has been stored in Model Registry. Here's the code for the pipeline:

@dlt.table

def trucklocation():

    return (

        spark.readStream

            .format("cloudFiles")

            .option("cloudFiles.format", "json")

            .option("cloudFiles.inferColumnTypes", True)

            .load(f"{source}/trucklocation")

            .select(

                F.current_timestamp().alias("processing_time"),

                "*"

            )

    )

loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

@dlt.table

def velocity_predictions():

    return (

        dlt.read("trucklocation")

            .withColumn('predictions', loaded_model_udf(struct(*map(col, ['velocity']))))

    )

The pipeline errors out with the following error:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 804.0 failed 4 times, most recent failure: Lost task 0.3 in stage 804.0 (TID 1285) (10.55.136.232 executor 0): org.apache.spark.api.python.PythonException: 'AttributeError: 'IsolationForest' object has no attribute 'n_features_''. Full traceback below:

Traceback (most recent call last):

File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-a35783aa-d900-4f51-9233-f8eb37babc87/lib/python3.9/site-packages/mlflow/pyfunc/__init__.py", line 1293, in udf

os.kill(scoring_server_proc.pid, signal.SIGTERM)

File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-a35783aa-d900-4f51-9233-f8eb37babc87/lib/python3.9/site-packages/mlflow/pyfunc/__init__.py", line 1080, in _predict_row_batch

result = predict_fn(pdf)

File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-a35783aa-d900-4f51-9233-f8eb37babc87/lib/python3.9/site-packages/mlflow/pyfunc/__init__.py", line 1274, in batch_predict_fn

return loaded_model.predict(pdf)

File "/local_disk0/.ephemeral_nfs/envs/pythonEnv-a35783aa-d900-4f51-9233-f8eb37babc87/lib/python3.9/site-packages/mlflow/pyfunc/__init__.py", line 427, in predict

return self._predict_fn(data)

File "/databricks/python/lib/python3.9/site-packages/sklearn/ensemble/_iforest.py", line 314, in predict

is_inlier[self.decision_function(X) < 0] = -1

File "/databricks/python/lib/python3.9/site-packages/sklearn/ensemble/_iforest.py", line 347, in decision_function

return self.score_samples(X) - self.offset_

File "/databricks/python/lib/python3.9/site-packages/sklearn/ensemble/_iforest.py", line 379, in score_samples

if self.n_features_ != X.shape[1]:

AttributeError: 'IsolationForest' object has no attribute 'n_features_'

I tried running the prediction directly as well this way and it worked fine:

loaded_model_udf = mlflow.pyfunc.spark_udf(spark, model_uri=model_uri)

df_location = spark.read.format('json').option("inferSchema", "true").load(s3path)

df = df_location.withColumn('predictions', loaded_model_udf(struct(*map(col, ['velocity']))))

Any help on why the pipeline would fail is appreciated.