code: def read_coco(file_path: str, split: str = "train"):
      """
      Read coco 2017 images into a SparkXShards using Spark backend.
  
      :param file_path: str. A HDFS path or local path of images.
      :param split: str. a split to read.
      :return: A new SparkXShards of tuple of image, target.
              target is a dictionary with a list of bbox, list of category_id, list of area,
              list of iscrouwd, list of segmentation.
              Each bbox contains four values in pixels [x_min, y_min, width, height].
      """
      spark = OrcaContext.get_spark_session()
      df = spark.read.json(
          file_path + "/annotations/instances_" + split + "2017.json"
      )
      ann_df = df.select(
          explode(col("annotations")).alias("annotations")
      )
      ann_df = ann_df.select(
          col("annotations.area").alias("area"),
          col("annotations.bbox").alias("bbox"),
          col("annotations.category_id").alias("category_id"),
          col("annotations.id").alias("id"),
          col("annotations.image_id").alias("image_id"),
          col("annotations.iscrowd").alias("iscrowd"),
          col("annotations.segmentation").alias("segmentation"),
      )
  
      ann_df = ann_df.groupby("image_id").agg(
          collect_list(col("bbox")).alias("bbox"),
          collect_list(col("category_id")).alias("category_id"),
          collect_list(col("area")).alias("area"),
          collect_list(col("iscrowd")).alias("iscrowd"),
          collect_list(col("segmentation")).alias("segmentation"),
      )
  
      ann_rdd = ann_df.rdd.map(
          lambda x: (
              x["image_id"],
              (
                  x["bbox"],
                  x["category_id"],
                  x["area"],
                  x["iscrowd"],
                  x["segmentation"],
              ),
          )
      )
  
      image_path = file_path + split + "2017/"
      file_names = get_file_paths(image_path)
      file_names = list(
          filter(
              lambda fn: fn.endswith(".jpg")
              and (not fn.startswith(".")),
              file_names,
          )
      )
      num_files = len(file_names)
      node_num, core_num = get_node_and_core_number()
      total_cores = node_num * core_num
      num_partitions = (
          num_files if num_files < total_cores else total_cores
      )
      # file_names is a rdd of(id, filename)
      file_names = spark.sparkContext.parallelize(
          file_names, num_partitions
      ).map(lambda x: (int(x.split("/")[-1].split(".")[0]), x))
  
      def load_image(iterator):
          for f in iterator:
              try:
                  img = open_image(f[1]).convert("RGB")
                  yield f[0], img
              except FileNotFoundError as e:
                  invalidOperationError(False, str(e), cause=e)
              yield f[0], None
  
      image_rdd = file_names.mapPartitions(load_image).filter(
          lambda x: x[1]
      )
  
      def transform_data(x):
          image = x[1][0]
          target = dict()
          target["bbox"] = x[1][1][0]
          target["category_id"] = x[1][1][1]
          target["area"] = x[1][1][2]
          target["iscrowd"] = x[1][1][3]
          target["segmentation"] = x[1][1][4]
          return image, target
  
      out_rdd = image_rdd.join(ann_rdd)
      out_rdd = out_rdd.map(lambda x: transform_data(x))
  
      return SparkXShards(out_rdd)
  
file_path: BigDL/python/orca/src/bigdl/orca/data/image/preprocessing.py
imports: from bigdl.orca import OrcaContext
  from typing import TYPE_CHECKING, Any, Tuple
  from bigdl.orca.data.file import *
  from bigdl.orca.data.utils import *
  from bigdl.orca.data import SparkXShards
  from pyspark.sql.functions import col, explode, collect_list
  from bigdl.dllib.utils.log4Error import invalidInputError
  from bigdl.dllib.utils.common import get_node_and_core_number
  import os.path
label_desc: Use image node_id 6 for calling a global function with example usage: read_coco(file_path, split) and returns: SparkXShards, image,  target