code: def main():
      anchors = yolo_anchors
      anchor_masks = yolo_anchor_masks
  
      parser = argparse.ArgumentParser()
      parser.add_argument(
          "--data_dir",
          dest="data_dir",
          help="Required. The path where data locates.",
      )
      parser.add_argument(
          "--output_data",
          dest="output_data",
          default=tempfile.mkdtemp(),
          help="Required. The path where voc parquet data locates.",
      )
      parser.add_argument(
          "--data_year",
          dest="data_year",
          default="2009",
          help="Required. The voc data date.",
      )
      parser.add_argument(
          "--split_name_train",
          dest="split_name_train",
          default="train",
          help="Required. Split name.",
      )
      parser.add_argument(
          "--split_name_test",
          dest="split_name_test",
          default="val",
          help="Required. Split name.",
      )
      parser.add_argument(
          "--names",
          dest="names",
          help="Required. The path where class names locates.",
      )
      parser.add_argument(
          "--weights",
          dest="weights",
          default="./checkpoints/yolov3.weights",
          help="Required. The path where weights locates.",
      )
      parser.add_argument(
          "--checkpoint",
          dest="checkpoint",
          default="./checkpoints/yolov3.tf",
          help="Required. The path where checkpoint locates.",
      )
      parser.add_argument(
          "--checkpoint_folder",
          dest="checkpoint_folder",
          default="./checkpoints",
          help="Required. The path where saved checkpoint locates.",
      )
      parser.add_argument(
          "--epochs",
          dest="epochs",
          type=int,
          default=2,
          help="Required. epochs.",
      )
      parser.add_argument(
          "--batch_size",
          dest="batch_size",
          type=int,
          default=16,
          help="Required. epochs.",
      )
      parser.add_argument(
          "--cluster_mode",
          dest="cluster_mode",
          default="local",
          help="Required. Run on local/yarn/k8s/spark-submit mode.",
      )
      parser.add_argument(
          "--class_num",
          dest="class_num",
          type=int,
          default=20,
          help="Required. class num.",
      )
      parser.add_argument(
          "--worker_num",
          type=int,
          default=1,
          help="The number of slave nodes to be used in the cluster."
          "You can change it depending on your own cluster setting.",
      )
      parser.add_argument(
          "--cores",
          type=int,
          default=4,
          help="The number of cpu cores you want to use on each node. "
          "You can change it depending on your own cluster setting.",
      )
      parser.add_argument(
          "--memory",
          type=str,
          default="20g",
          help="The memory you want to use on each node. "
          "You can change it depending on your own cluster setting.",
      )
      parser.add_argument(
          "--object_store_memory",
          type=str,
          default="10g",
          help="The memory you want to use on each node. "
          "You can change it depending on your own cluster setting.",
      )
      parser.add_argument(
          "--enable_numa_binding",
          dest="enable_numa_binding",
          default=False,
          help="enable_numa_binding",
      )
      parser.add_argument(
          "--k8s_master",
          type=str,
          default="",
          help="The k8s master. "
          "It should be k8s://https://<k8s-apiserver-host>: "
          "<k8s-apiserver-port>.",
      )
      parser.add_argument(
          "--container_image",
          type=str,
          default="",
          help="The runtime k8s image. ",
      )
      parser.add_argument(
          "--k8s_driver_host",
          type=str,
          default="",
          help="The k8s driver localhost.",
      )
      parser.add_argument(
          "--k8s_driver_port",
          type=str,
          default="",
          help="The k8s driver port.",
      )
      parser.add_argument(
          "--nfs_mount_path",
          type=str,
          default="",
          help="nfs mount path",
      )
  
      options = parser.parse_args()
  
      if options.cluster_mode == "local":
          init_orca_context(
              cluster_mode="local",
              cores=options.cores,
              num_nodes=options.worker_num,
              memory=options.memory,
              init_ray_on_spark=True,
              object_store_memory=options.object_store_memory,
          )
      elif options.cluster_mode == "k8s":
          init_orca_context(
              cluster_mode="k8s",
              master=options.k8s_master,
              container_image=options.container_image,
              init_ray_on_spark=True,
              enable_numa_binding=options.enable_numa_binding,
              num_nodes=options.worker_num,
              cores=options.cores,
              memory=options.memory,
              object_store_memory=options.object_store_memory,
              conf={
                  "spark.driver.host": options.driver_host,
                  "spark.driver.port": options.driver_port,
                  "spark.kubernetes.executor.volumes.persistentVolumeClaim."
                  "nfsvolumeclaim.options.claimName": "nfsvolumeclaim",
                  "spark.kubernetes.executor.volumes.persistentVolumeClaim."
                  "nfsvolumeclaim.mount.path": options.nfs_mount_path,
                  "spark.kubernetes.driver.volumes.persistentVolumeClaim."
                  "nfsvolumeclaim.options.claimName": "nfsvolumeclaim",
                  "spark.kubernetes.driver.volumes.persistentVolumeClaim."
                  "nfsvolumeclaim.mount.path": options.nfs_mount_path,
              },
          )
      elif options.cluster_mode.startswith("yarn"):
          init_orca_context(
              cluster_mode=options.cluster_mode,
              cores=options.cores,
              num_nodes=options.worker_num,
              memory=options.memory,
              init_ray_on_spark=True,
              enable_numa_binding=options.enable_numa_binding,
              object_store_memory=options.object_store_memory,
          )
      elif options.cluster_mode == "spark-submit":
          init_orca_context(cluster_mode="spark-submit")
      # convert yolov3 weights
      yolo = YoloV3(classes=80)
      load_darknet_weights(yolo, options.weights)
      yolo.save_weights(options.checkpoint)
  
      def model_creator(config):
          model = YoloV3(
              DEFAULT_IMAGE_SIZE,
              training=True,
              classes=options.class_num,
          )
          anchors = yolo_anchors
          anchor_masks = yolo_anchor_masks
  
          model_pretrained = YoloV3(
              DEFAULT_IMAGE_SIZE, training=True, classes=80
          )
          model_pretrained.load_weights(options.checkpoint)
  
          model.get_layer("yolo_darknet").set_weights(
              model_pretrained.get_layer("yolo_darknet").get_weights()
          )
          freeze_all(model.get_layer("yolo_darknet"))
  
          optimizer = tf.keras.optimizers.Adam(lr=1e-3)
          loss = [
              YoloLoss(anchors[mask], classes=options.class_num)
              for mask in anchor_masks
          ]
          model.compile(
              optimizer=optimizer, loss=loss, run_eagerly=False
          )
          return model
  
      # prepare data
      with open(options.names) as f:
          class_map = {
              name: idx
              for idx, name in enumerate(f.read().splitlines())
          }
      dataset_path = os.path.join(options.data_dir, "VOCdevkit")
      voc_train_path = os.path.join(
          options.output_data, "train_dataset"
      )
      voc_val_path = os.path.join(options.output_data, "val_dataset")
  
      write_parquet(
          format="voc",
          voc_root_path=dataset_path,
          output_path="file://" + voc_train_path,
          splits_names=[(options.data_year, options.split_name_train)],
          classes=class_map,
      )
      write_parquet(
          format="voc",
          voc_root_path=dataset_path,
          output_path="file://" + voc_val_path,
          splits_names=[(options.data_year, options.split_name_test)],
          classes=class_map,
      )
  
      output_types = {
          "image": tf.string,
          "label": tf.float32,
          "image_id": tf.string,
      }
      output_shapes = {"image": (), "label": (None, 5), "image_id": ()}
  
      def train_data_creator(config, batch_size):
          train_dataset = read_parquet(
              format="tf_dataset",
              path=voc_train_path,
              output_types=output_types,
              output_shapes=output_shapes,
          )
          train_dataset = train_dataset.map(
              lambda data_dict: (data_dict["image"], data_dict["label"])
          )
          train_dataset = train_dataset.map(parse_data_train)
          train_dataset = train_dataset.shuffle(buffer_size=512)
          train_dataset = train_dataset.batch(batch_size)
          train_dataset = train_dataset.map(
              lambda x, y: (
                  transform_images(x, DEFAULT_IMAGE_SIZE),
                  transform_targets(
                      y, anchors, anchor_masks, DEFAULT_IMAGE_SIZE
                  ),
              )
          )
          train_dataset = train_dataset.prefetch(
              buffer_size=tf.data.experimental.AUTOTUNE
          )
          return train_dataset
  
      def val_data_creator(config, batch_size):
          val_dataset = read_parquet(
              format="tf_dataset",
              path=voc_val_path,
              output_types=output_types,
              output_shapes=output_shapes,
          )
          val_dataset = val_dataset.map(
              lambda data_dict: (data_dict["image"], data_dict["label"])
          )
          val_dataset = val_dataset.map(parse_data_train)
          val_dataset = val_dataset.batch(batch_size)
          val_dataset = val_dataset.map(
              lambda x, y: (
                  transform_images(x, DEFAULT_IMAGE_SIZE),
                  transform_targets(
                      y, anchors, anchor_masks, DEFAULT_IMAGE_SIZE
                  ),
              )
          )
          return val_dataset
  
      callbacks = [
          ReduceLROnPlateau(verbose=1),
          EarlyStopping(patience=3, verbose=1),
          ModelCheckpoint(
              options.checkpoint_folder + "/yolov3_train_{epoch}.tf",
              verbose=1,
              save_weights_only=True,
          ),
          TensorBoard(log_dir="logs"),
      ]
  
      trainer = Estimator.from_keras(model_creator=model_creator)
  
      trainer.fit(
          train_data_creator,
          epochs=options.epochs,
          batch_size=options.batch_size,
          callbacks=callbacks,
          validation_data=val_data_creator,
      )
      stop_orca_context()
  
imports: import tensorflow
  from tensorflow.keras.callbacks import ReduceLROnPlateau, EarlyStopping, ModelCheckpoint, TensorBoard
  from tensorflow.keras import Model
  from tensorflow.keras.layers import Add, Concatenate, Conv2D, Input, Lambda, LeakyReLU, MaxPool2D, UpSampling2D, ZeroPadding2D, BatchNormalization
  from tensorflow.keras.regularizers import l2
  from tensorflow.keras.losses import binary_crossentropy, sparse_categorical_crossentropy
  from bigdl.dllib.utils.log4Error import invalidInputError
  from bigdl.orca.data.image.parquet_dataset import read_parquet, write_parquet
  from bigdl.orca.learn.tf2 import Estimator
  from bigdl.orca import init_orca_context, stop_orca_context
  import numpy
  import tempfile
  import os
  import argparse
label_desc: Use image node_id 19 for calling a global function with example usage: main() and returns: model, train_dataset, val_dataset