code: def read_data(data_dir, dataset):
      spark = OrcaContext.get_spark_session()
      print("Loading data...")
  
      schema = StructType(
          [
              StructField("user", LongType(), False),
              StructField("item", LongType(), False),
          ]
      )
      if dataset == "ml-1m":
          schema_user = StructType(
              [
                  StructField("user", LongType(), False),
                  StructField("gender", StringType(), False),
                  StructField("age", IntegerType(), False),
                  StructField("occupation", StringType(), False),
                  StructField("zipcode", StringType(), False),
              ]
          )
          schema_item = StructType(
              [
                  StructField("item", LongType(), False),
                  StructField("title", StringType(), False),
                  StructField("category", StringType(), False),
              ]
          )
          # Need spark3 to support delimiter with more than one character.
          df_rating = spark.read.csv(
              os.path.join(data_dir, dataset, "ratings.dat"),
              sep="::",
              schema=schema,
              header=False,
          )
          df_user = spark.read.csv(
              os.path.join(data_dir, dataset, "users.dat"),
              sep="::",
              schema=schema_user,
              header=False,
          )
          df_item = spark.read.csv(
              os.path.join(data_dir, dataset, "movies.dat"),
              sep="::",
              schema=schema_item,
              header=False,
          )
      else:  # ml-100k
          schema_user = StructType(
              [
                  StructField("user", LongType(), False),
                  StructField("age", IntegerType(), False),
                  StructField("gender", StringType(), False),
                  StructField("occupation", StringType(), False),
                  StructField("zipcode", StringType(), False),
              ]
          )
          schema_item = StructType(
              [
                  StructField("item", LongType(), False),
                  StructField("title", StringType(), False),
                  StructField("date", StringType(), False),
                  StructField("vdate", StringType(), False),
                  StructField("URL", StringType(), False),
              ]
              + [
                  StructField(f"col{i}", StringType(), False)
                  for i in range(19)
              ]
          )
          df_rating = spark.read.csv(
              os.path.join(data_dir, dataset, "u.data"),
              sep="\t",
              schema=schema,
              header=False,
          )
          df_user = spark.read.csv(
              os.path.join(data_dir, dataset, "u.user"),
              sep="|",
              schema=schema_user,
              header=False,
          )
          df_item = spark.read.csv(
              os.path.join(data_dir, dataset, "u.item"),
              sep="|",
              schema=schema_item,
              header=False,
          )
          # Merge multiple one-hot columns into one movie category column
          df_item = df_item.select(
              df_item.item,
              concat(*[df_item[f"col{i}"] for i in range(19)]).alias(
                  "category"
              ),
          )
      return df_rating, df_user, df_item
  
file_path: BigDL/python/orca/tutorial/NCF/process_spark_dataframe.py
imports: import os
  from pyspark.ml.feature import StringIndexer, MinMaxScaler, VectorAssembler
  from pyspark.ml import Pipeline
  from pyspark.sql.types import StructField, StructType, IntegerType, LongType, ArrayType, StringType
  from pyspark.sql.functions import udf, lit, collect_list, explode, concat
  from bigdl.orca import OrcaContext
label_desc: Use image node_id 1 for calling a global function with example usage: read_data(data_dir, dataset) and returns: df_rating,  df_user,  df_item