Personal Cluster 15.3 ML, Running the following notebook:

import pyspark.sql.functions as F
from datetime import datetime, timedelta

spark.sparkContext.addPyFile("gs://spark-lib/bigquery/spark-bigquery-support-0.26.0.zip")

target_hour = datetime(2024, 7, 24, 15)  # Year, Month, Day, Hour
start_time = target_hour.strftime("%Y-%m-%d %H:%M:%S")
end_time = (target_hour + timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")

df = spark.read.format("bigquery") \
  .option("project", "e24-prod-data") \
  .option("parentProject", "databricks-424220") \
  .option("table", input_table_name) \
  .option("filter", f"date_ingested >= '{start_time}' AND date_ingested < '{end_time}'") \
  .load() \
  .withColumns( {
            "date": F.date_format(F.col("date_ingested"), "yyyy-MM-dd"),
            "hour": F.date_format(F.col("date_ingested"), "HH"),
        }) \
  .load()
display(df)

I get the following error

AssertionError: Undefined error message parameter for error class: CANNOT_PARSE_DATATYPE. Parameters: {'error': "No module named 'google.cloud.spark'"}
File /databricks/spark/python/pyspark/sql/dataframe.py:657, in DataFrame.schema(self)
    655 try:
    656     self._schema = cast(
--> 657         StructType, _parse_datatype_json_string(self._jdf.schema().json())
    658     )
    659 except Exception as e:
File /databricks/spark/python/pyspark/sql/types.py:1817, in _parse_datatype_json_string(json_string)
   1763 """Parses the given data type JSON string.
   1764 
   1765 Examples
   (...)
   1815 >>> check_datatype(complex_maptype)
   1816 """
-> 1817 return _parse_datatype_json_value(json.loads(json_string))
File /databricks/spark/python/pyspark/sql/types.py:1877, in _parse_datatype_json_value(json_value, fieldPath, collationsMap)
   1876         return MapType.fromJson(json_value, fieldPath, collationsMap)
-> 1877     return StructType.fromJson(json_value)
   1878 elif tpe == "udt":
File /databricks/spark/python/pyspark/sql/types.py:1361, in StructType.fromJson(cls, json)
   1270 """
   1271 Constructs :class:`StructType` from a schema defined in JSON format.
   1272 
   (...)
   1359 'struct<Person:struct<name:string,surname:string>>'
   1360 """
-> 1361 return StructType([StructField.fromJson(f) for f in json["fields"]])
File /databricks/spark/python/pyspark/sql/types.py:1361, in <listcomp>(.0)
   1270 """
   1271 Constructs :class:`StructType` from a schema defined in JSON format.
   1272 
   (...)
   1359 'struct<Person:struct<name:string,surname:string>>'
   1360 """
-> 1361 return StructType([StructField.fromJson(f) for f in json["fields"]])
File /databricks/spark/python/pyspark/sql/types.py:954, in StructField.fromJson(cls, json)
    948     metadata = {
    949         key: value for key, value in metadata.items() if key != _COLLATIONS_METADATA_KEY
    950     }
    952 return StructField(
    953     json["name"],
--> 954     _parse_datatype_json_value(json["type"], json["name"], collationsMap),
    955     json.get("nullable", True),
    956     metadata,
    957 )
File /databricks/spark/python/pyspark/sql/types.py:1877, in _parse_datatype_json_value(json_value, fieldPath, collationsMap)
   1876         return MapType.fromJson(json_value, fieldPath, collationsMap)
-> 1877     return StructType.fromJson(json_value)
   1878 elif tpe == "udt":
File /databricks/spark/python/pyspark/sql/types.py:1361, in StructType.fromJson(cls, json)
   1270 """
   1271 Constructs :class:`StructType` from a schema defined in JSON format.
   1272 
   (...)
   1359 'struct<Person:struct<name:string,surname:string>>'
   1360 """
-> 1361 return StructType([StructField.fromJson(f) for f in json["fields"]])
File /databricks/spark/python/pyspark/sql/types.py:1361, in <listcomp>(.0)
   1270 """
   1271 Constructs :class:`StructType` from a schema defined in JSON format.
   1272 
   (...)
   1359 'struct<Person:struct<name:string,surname:string>>'
   1360 """
-> 1361 return StructType([StructField.fromJson(f) for f in json["fields"]])
File /databricks/spark/python/pyspark/sql/types.py:954, in StructField.fromJson(cls, json)
    948     metadata = {
    949         key: value for key, value in metadata.items() if key != _COLLATIONS_METADATA_KEY
    950     }
    952 return StructField(
    953     json["name"],
--> 954     _parse_datatype_json_value(json["type"], json["name"], collationsMap),
    955     json.get("nullable", True),
    956     metadata,
    957 )
File /databricks/spark/python/pyspark/sql/types.py:1879, in _parse_datatype_json_value(json_value, fieldPath, collationsMap)
   1878 elif tpe == "udt":
-> 1879     return UserDefinedType.fromJson(json_value)
   1880 else:
File /databricks/spark/python/pyspark/sql/types.py:1565, in UserDefinedType.fromJson(cls, json)
   1564 pyClass = pyUDT[split + 1 :]
-> 1565 m = __import__(pyModule, globals(), locals(), [pyClass])
   1566 if not hasattr(m, pyClass):
ModuleNotFoundError: No module named 'google.cloud.spark'

During handling of the above exception, another exception occurred:
AssertionError                            Traceback (most recent call last)
File <command-2993807820048799>, line 26
     10 # use with .option("query", query) instead of option table and filter
     11 query = f"""SELECT *
     12 FROM ev24_ep_data_warehouse_v1_prd.metabase_news__2020_2029__common_v1__source_text_common_fields
     13 WHERE date_ingested BETWEEN '{start_time}' '{end_time}'
     14 """
     16 df = spark.read.format("bigquery") \
     17   .option("project", "e24-prod-data") \
     18   .option("parentProject", "databricks-424220") \
     19   .option("table", input_table_name) \
     20   .option("filter", f"date_ingested >= '{start_time}' AND date_ingested < '{end_time}'") \
     21   .load() \
     22   .withColumns( {
     23             "date": F.date_format(F.col("date_ingested"), "yyyy-MM-dd"),
     24             "hour": F.date_format(F.col("date_ingested"), "HH"),
     25         }) \
---> 26   .load()
     27 display(df)
File /databricks/spark/python/pyspark/instrumentation_utils.py:47, in _wrap_function.<locals>.wrapper(*args, **kwargs)
     45 start = time.perf_counter()
     46 try:
---> 47     res = func(*args, **kwargs)
     48     logger.log_success(
     49         module_name, class_name, function_name, time.perf_counter() - start, signature
     50     )
     51     return res
File /databricks/spark/python/pyspark/sql/dataframe.py:3745, in DataFrame.__getattr__(self, name)
   3712 def __getattr__(self, name: str) -> Column:
   3713     """Returns the :class:`Column` denoted by ``name``.
   3714 
   3715     .. versionadded:: 1.3.0
   (...)
   3743     +---+
   3744     """
-> 3745     if name not in self.columns:
   3746         raise PySparkAttributeError(
   3747             error_class="ATTRIBUTE_NOT_SUPPORTED", message_parameters={"attr_name": name}
   3748         )
   3749     jc = self._jdf.apply(name)
File /databricks/spark/python/pyspark/instrumentation_utils.py:75, in _wrap_property.<locals>.wrapper(self)
     71 @property  # type: ignore[misc]
     72 def wrapper(self: Any) -> Any:
     73     if hasattr(_local, "logging") and _local.logging:
     74         # no need to log since this should be internal call.
---> 75         return prop.fget(self)
     76     _local.logging = True
     77     try:
File /databricks/spark/python/pyspark/sql/dataframe.py:2592, in DataFrame.columns(self)
   2517 @property
   2518 def columns(self) -> List[str]:
   2519     """
   2520     Retrieves the names of all columns in the :class:`DataFrame` as a list.
   2521 
   (...)
   2590     False
   2591     """
-> 2592     return [f.name for f in self.schema.fields]
File /databricks/spark/python/pyspark/instrumentation_utils.py:75, in _wrap_property.<locals>.wrapper(self)
     71 @property  # type: ignore[misc]
     72 def wrapper(self: Any) -> Any:
     73     if hasattr(_local, "logging") and _local.logging:
     74         # no need to log since this should be internal call.
---> 75         return prop.fget(self)
     76     _local.logging = True
     77     try:
File /databricks/spark/python/pyspark/sql/dataframe.py:660, in DataFrame.schema(self)
    656         self._schema = cast(
    657             StructType, _parse_datatype_json_string(self._jdf.schema().json())
    658         )
    659     except Exception as e:
--> 660         raise PySparkValueError(
    661             error_class="CANNOT_PARSE_DATATYPE",
    662             message_parameters={"error": str(e)},
    663         )
    664 return self._schema
File /databricks/spark/python/pyspark/errors/exceptions/base.py:45, in PySparkException.__init__(self, message, error_class, message_parameters, query_contexts)
     42 self._error_reader = ErrorClassesReader()
     44 if message is None:
---> 45     self._message = self._error_reader.get_error_message(
     46         cast(str, error_class), cast(Dict[str, str], message_parameters)
     47     )
     48 else:
     49     self._message = message
File /databricks/spark/python/pyspark/errors/utils.py:39, in ErrorClassesReader.get_error_message(self, error_class, message_parameters)
     37 # Verify message parameters.
     38 message_parameters_from_template = re.findall("<([a-zA-Z0-9_-]+)>", message_template)
---> 39 assert set(message_parameters_from_template) == set(message_parameters), (
     40     f"Undefined error message parameter for error class: {error_class}. "
     41     f"Parameters: {message_parameters}"
     42 )
     44 def replace_match(match: Match[str]) -> str:
     45     return match.group().translate(str.maketrans("<>", "{}"))

Any idea how to resolve this?