code: def search(args):
      set_env(args.num_threads)
      if args.cluster_mode == "yarn":
          sc = init_orca_context(
              "yarn",
              cores=args.num_threads,
              num_nodes=args.num_tasks,
              memory=args.memory,
              extra_python_lib="utils.py",
          )
  
          print("add files to spark >>>>>>")
          sc.addFile(args.dict_path)
          sc.addFile(args.faiss_index_path)
  
          args.faiss_index_path = args.faiss_index_path.split("/")[-1]
          args.dict_path = args.dict_path.split("/")[-1]
      elif args.cluster_mode == "spark-submit":
          sc = init_orca_context("spark-submit")
      elif args.cluster_mode == "local":
          sc = init_orca_context(
              "local", cores=args.num_threads * args.num_tasks
          )
      else:
          invalidInputError(
              False,
              "cluster_mode should be one of 'local', 'spark-submit' and "
              "'yarn', but got " + args.cluster_mode,
          )
          sys.exit()
  
      spark = OrcaContext.get_spark_session()
      with StopWatch("do_search spark >>>>>>") as sw:
          df = spark.read.parquet(args.parquet_path)
          print("Total number of items: ", df.count())
          rdd = df.rdd.repartition(
              args.num_tasks
          )  # Each node runs one faiss task
          res_rdd = rdd.mapPartitions(
              faiss_search(
                  args.faiss_index_path,
                  args.dict_path,
                  args.cluster_mode,
                  batch_size=args.batch_size,
                  k=args.top_k,
              )
          )
          schema = StructType(
              [
                  StructField("seed_item", StringType(), False),
                  StructField("rec_item", StringType(), False),
                  StructField("rank", IntegerType(), False),
                  StructField("similarity_score", FloatType(), False),
              ]
          )
          res_df = spark.createDataFrame(res_rdd, schema=schema)
          res_df.write.mode("overwrite").parquet(
              args.parquet_output_path
          )
          stop_orca_context()
  
file_path: BigDL/python/friesian/example/recall/search.py
imports: import sys
  import argparse
  import numpy
  from bigdl.dllib.utils.log4Error import *
  from bigdl.orca import init_orca_context, OrcaContext, stop_orca_context
  from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType
  from utils import *
label_desc: Use image node_id 2 for calling a global function with example usage: search(args) without return types