code: def read_images_pil(
      file_path: str,
      label_func: Callable = None,
      target_path: str = None,
      image_type: str = ".jpg",
      target_type: str = ".png",
  ) -> "SparkXShards":
      """
      Read images into a SparkXShards using PIL.
  
      :param file_path: str. A HDFS path or local path of images.
      :param label_func: Callable. A function to get label from filename. Default is None.
      :param target_path: str. A HDFS path or local path of target images.
      :param image_type: str. Suffix of images.
      :param target_type: str. Suffix of target images.
      :return: A new SparkXShards of tuple of image, target.
      """
      sc = OrcaContext.get_spark_context()
  
      img_paths = get_file_paths(file_path)
      img_paths = list(
          filter(
              lambda fn: fn.endswith(image_type)
              and (not fn.startswith(".")),
              img_paths,
          )
      )
      num_files = len(img_paths)
      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
      )
  
      rdd = sc.parallelize(sorted(img_paths), num_partitions)
  
      def load_image(iterator):
          for f in iterator:
              img = open_image(f)
              if label_func:
                  img = img, label_func(f)
              yield img
  
      image_rdd = rdd.mapPartitions(load_image)
  
      if target_path:
          invalidInputError(
              label_func is None,
              "label_func should be None when reading targets directly",
          )
          target_paths = get_file_paths(target_path)
          target_paths = list(
              filter(
                  lambda fn: fn.endswith(target_type)
                  and (not fn.startswith(".")),
                  target_paths,
              )
          )
  
          invalidInputError(
              len(target_paths) == len(img_paths),
              "number of target files and image files should be the same",
          )
  
          t_rdd = sc.parallelize(sorted(target_paths), num_partitions)
          target_rdd = t_rdd.mapPartitions(load_image)
          image_rdd = image_rdd.zip(target_rdd)
  
      return SparkXShards(image_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 3 for calling a global function with example usage: read_images_pil(file_path, label_func, target_path, image_type, target_type) and returns: SparkXShards