code: def load_spark_df(
      spark,
      size="100k",
      header=None,
      schema=None,
      local_cache_path=None,
      dbutils=None,
      title_col=None,
      genres_col=None,
      year_col=None,
  ):
      """Loads the MovieLens dataset as `pyspark.sql.DataFrame`.
  
      Download the dataset from https://files.grouplens.org/datasets/movielens, unzip, and load as `pyspark.sql.DataFrame`.
  
      To load movie information only, you can use `load_item_df` function.
  
      Args:
          spark (pyspark.SparkSession): Spark session.
          size (str): Size of the data to load. One of ("100k", "1m", "10m", "20m", "mock100").
          header (list or tuple): Rating dataset header.
              If `schema` is provided or `size` is set to any of 'MOCK_DATA_FORMAT', this argument is ignored.
          schema (pyspark.StructType): Dataset schema.
              If `size` is set to any of 'MOCK_DATA_FORMAT', data is rendered in the 'MockMovielensSchema' instead.
          local_cache_path (str): Path (directory or a zip file) to cache the downloaded zip file.
              If None, all the intermediate files will be stored in a temporary directory and removed after use.
          dbutils (Databricks.dbutils): Databricks utility object
              If `size` is set to any of 'MOCK_DATA_FORMAT', this parameter is ignored.
          title_col (str): Title column name. If None, the column will not be loaded.
          genres_col (str): Genres column name. Genres are '|' separated string.
              If None, the column will not be loaded.
          year_col (str): Movie release year column name. If None, the column will not be loaded.
              If `size` is set to any of 'MOCK_DATA_FORMAT', this parameter is ignored.
  
      Returns:
          pyspark.sql.DataFrame: Movie rating dataset.
  
      **Examples**
  
      .. code-block:: python
  
          # To load just user-id, item-id, and ratings from MovieLens-1M dataset:
          spark_df = load_spark_df(spark, '1m', ('UserId', 'ItemId', 'Rating'))
  
          # The schema can be defined as well:
          schema = StructType([
              StructField(DEFAULT_USER_COL, IntegerType()),
              StructField(DEFAULT_ITEM_COL, IntegerType()),
              StructField(DEFAULT_RATING_COL, FloatType()),
              StructField(DEFAULT_TIMESTAMP_COL, LongType()),
              ])
          spark_df = load_spark_df(spark, '1m', ('UserId', 'ItemId', 'Rating'), schema=schema)
  
          # To load rating's timestamp together:
          spark_df = load_spark_df(spark, '1m', ('UserId', 'ItemId', 'Rating', 'Timestamp'))
  
          # To load movie's title, genres, and released year info along with the ratings data:
          spark_df = load_spark_df(spark, '1m', ('UserId', 'ItemId', 'Rating', 'Timestamp'),
              title_col='Title',
              genres_col='Genres',
              year_col='Year'
          )
  
          # On DataBricks, pass the dbutils argument as follows:
          spark_df = load_spark_df(spark, dbutils=dbutils)
      """
      size = size.lower()
      if size not in DATA_FORMAT and size not in MOCK_DATA_FORMAT:
          raise ValueError(f"Size: {size}. " + ERROR_MOVIE_LENS_SIZE)
  
      if size in MOCK_DATA_FORMAT:
          # generate fake data
          return MockMovielensSchema.get_spark_df(
              spark,
              keep_title_col=(title_col is not None),
              keep_genre_col=(genres_col is not None),
              **MOCK_DATA_FORMAT[
                  size
              ],  # supply the rest of the kwarg with the dictionary
          )
  
      schema = _get_schema(header, schema)
      if len(schema) < 2:
          raise ValueError(ERROR_HEADER)
  
      movie_col = schema[1].name
  
      with download_path(local_cache_path) as path:
          filepath = os.path.join(path, "ml-{}.zip".format(size))
          datapath, item_datapath = _maybe_download_and_extract(
              size, filepath
          )
          spark_datapath = (
              "file:///" + datapath
          )  # shorten form of file://localhost/
  
          # Load movie features such as title, genres, and release year.
          # Since the file size is small, we directly load as pd.DataFrame from the driver node
          # and then convert into pyspark.sql.DataFrame
          item_pd_df = _load_item_df(
              size,
              item_datapath,
              movie_col,
              title_col,
              genres_col,
              year_col,
          )
          item_df = (
              spark.createDataFrame(item_pd_df)
              if item_pd_df is not None
              else None
          )
  
          if is_databricks():
              if dbutils is None:
                  raise ValueError(
                      """
                      To use on a Databricks, dbutils object should be passed as an argument.
                      E.g. load_spark_df(spark, dbutils=dbutils)
                  """
                  )
  
              # Move rating file to DBFS in order to load into pyspark.sql.DataFrame
              dbfs_datapath = "dbfs:/tmp/" + datapath
              dbutils.fs.mv(spark_datapath, dbfs_datapath)
              spark_datapath = dbfs_datapath
  
          # pyspark's read csv currently doesn't support multi-character delimiter, thus we manually handle that
          separator = DATA_FORMAT[size].separator
          if len(separator) > 1:
              raw_data = spark.sparkContext.textFile(spark_datapath)
              data_rdd = raw_data.map(lambda l: l.split(separator)).map(
                  lambda c: [
                      int(c[0]),
                      int(c[1]),
                      float(c[2]),
                      int(c[3]),
                  ][: len(schema)]
              )
              df = spark.createDataFrame(data_rdd, schema)
          else:
              df = spark.read.csv(
                  spark_datapath,
                  schema=schema,
                  sep=separator,
                  header=DATA_FORMAT[size].has_header,
              )
  
          # Merge rating df w/ item_df
          if item_df is not None:
              df = df.join(item_df, movie_col, "left")
  
          # Cache and force trigger action since data-file might be removed.
          df.cache()
          df.count()
  
      return df
  
imports: import os
  import re
  import random
  import shutil
  import warnings
  import pandas
  from typing import Optional
  from zipfile import ZipFile
  from recommenders.datasets.download_utils import maybe_download, download_path
  from recommenders.utils.notebook_utils import is_databricks
  from recommenders.utils.constants import DEFAULT_HEADER, DEFAULT_ITEM_COL, DEFAULT_USER_COL, DEFAULT_RATING_COL, DEFAULT_TIMESTAMP_COL, DEFAULT_TITLE_COL, DEFAULT_GENRE_COL
  import pandera
  import pandera.extensions
  from pandera import Field
  from pandera.typing import Series
label_desc: Use image node_id 4 for calling a global function with example usage: load_spark_df(spark, size, header, schema, local_cache_path, dbutils, title_col, genres_col, year_col) and returns: df, MockMovielensSchema