code: def prepare_data(data_dir="./", dataset="ml-1m", neg_scale=4):
      df_rating, df_user, df_item = read_data(data_dir, dataset)
  
      user_num = (
          df_rating.agg({"user": "max"}).collect()[0]["max(user)"] + 1
      )
      item_num = (
          df_rating.agg({"item": "max"}).collect()[0]["max(item)"] + 1
      )
  
      df_rating = generate_neg_sample(
          df_rating, item_num, neg_scale=neg_scale
      )
      df, sparse_feats_input_dims = merge_features(
          df_rating, df_user, df_item, sparse_features, dense_features
      )
  
      train_df, val_df = df.randomSplit([0.8, 0.2], seed=100)
      return (
          train_df,
          val_df,
          user_num,
          item_num,
          sparse_feats_input_dims,
          len(dense_features),
          get_feature_cols(),
          get_label_cols(),
      )
  
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 6 for calling a global function with example usage: prepare_data(data_dir, dataset, neg_scale) and returns: train_df,  val_df,  user_num,  item_num,  sparse_feats_input_dims,  len,  get_feature_cols,  get_label_cols