code: def generate_neg_sample(df, item_num, neg_scale):
      def neg_sample(x):
          import random
  
          neg_res = []
          for _ in x:
              for i in range(neg_scale):
                  neg_item_index = random.randint(1, item_num - 1)
                  while neg_item_index in x:
                      neg_item_index = random.randint(1, item_num - 1)
                  neg_res.append(neg_item_index)
          return neg_res
  
      df = df.withColumn("label", lit(1.0))
  
      neg_sample_udf = udf(neg_sample, ArrayType(LongType(), False))
      df_neg = df.groupBy("user").agg(
          neg_sample_udf(collect_list("item")).alias("item_list")
      )
      df_neg = df_neg.select(df_neg.user, explode(df_neg.item_list))
      df_neg = df_neg.withColumn("label", lit(0.0))
      df_neg = df_neg.withColumnRenamed("col", "item")
  
      df = df.unionByName(df_neg)
      df = df.repartition(df.rdd.getNumPartitions())
      return df
  
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 2 for calling a global function with example usage: generate_neg_sample(df, item_num, neg_scale) and returns: df, neg_res