Help with Presidio with PySpark on Databricks+AWS

[advait-dixit]

I am trying to run Presidio in a notebook on Databricks using this example: https://microsoft.github.io/presidio/samples/deployments/spark/.

Here is my code which is very similar to the code in the example:

from presidio_analyzer import AnalyzerEngine
from presidio_analyzer.nlp_engine import NlpEngineProvider
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig
from pyspark.sql.types import StringType
import pyspark.pandas
from databricks.connect import DatabricksSession
from pyspark.sql.functions import broadcast
from pyspark.sql.functions import col, pandas_udf
import pandas as pd

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

broadcasted_analyzer = sc.broadcast(analyzer)
broadcasted_anonymizer = sc.broadcast(anonymizer)

def anonymize_text(text: str) -> str:
    analyzer_results = broadcasted_analyzer.value.analyze(text=text, language="en")
    anonymized_results = broadcasted_anonymizer.value.anonymize(
        text=text,
        analyzer_results=analyzer_results,
        operators={"DEFAULT": OperatorConfig("replace", {"new_value": "<ANONYMIZED>"})},
    )
    return anonymized_results.text


def anonymize_series(s: pd.Series) -> pd.Series:
    return s.apply(anonymize_text)

# define a the function as pandas UDF
anonymize = pandas_udf(anonymize_series, returnType=StringType())

anonymized_column = "col1"
df = spark.read.table("`test-catalog`.default.table1")

# apply the udf
anonymized_df = df.withColumn(
    anonymized_column+"_classification", anonymize(col(anonymized_column))
)
anonymized_df.write.mode("overwrite").option("mergeSchema", "true").saveAsTable(f"`test-catalog`.default.anonymized_table1")

However, my notebook fails with the following error at line 41 (last line) of above code:

  An exception was thrown from the Python worker. Please see the stack trace below.
Traceback (most recent call last):
  File "/databricks/spark/python/pyspark/serializers.py", line 192, in _read_with_length
    return self.loads(obj)
           ^^^^^^^^^^^^^^^
  File "/databricks/spark/python/pyspark/serializers.py", line 572, in loads
    return cloudpickle.loads(obj, encoding=encoding)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/databricks/spark/python/pyspark/core/broadcast.py", line 61, in _from_id
    raise PySparkRuntimeError(
pyspark.errors.exceptions.base.PySparkRuntimeError: [BROADCAST_VARIABLE_NOT_LOADED] Broadcast variable `19` not loaded.

[advait-dixit]

I have also installed this following libraries in the "Libraries" tab of the cluster:

Name	Type
en_core_web_lg-3.7.1-py3-none-any.whl	Wheel
presidio-analyzer==2.2.354	PyPI
presidio-anonymizer==2.2.354	PyPI