code: def fit_ray_xshards(self, train_shards, val_shards):
      if val_shards is None:
          # TODO: split data_creator from step
          def transform_func(worker, partition_refs):
              return worker.step.remote(partition_refs)
  
          worker_stats = train_shards.reduce_partitions_for_actors(
              self.remote_workers, transform_func
          )
      else:
  
          def zip_func(
              worker, this_partition_refs, that_partition_refs
          ):
              return worker.step.remote(
                  this_partition_refs, that_partition_refs
              )
  
          worker_stats = train_shards.zip_reduce_shards_with_actors(
              val_shards, self.remote_workers, zip_func
          )
      return worker_stats
  
imports: import time
  import argparse
  import tensorflow
  import os
  import sys
  import math
  import collections
  from tensorflow.python.client import timeline
  import json
  from tensorflow.python.ops import partitioned_variables
  import ray
  from bigdl.dllib.utils.log4Error import *
  from bigdl.orca import init_orca_context, stop_orca_context, OrcaContext
  from bigdl.orca.data.utils import process_spark_xshards
  from bigdl.orca.learn.utils import maybe_dataframe_to_xshards
  from bigdl.friesian.feature import FeatureTable
label_desc: Use image node_id 3 for calling the RayDeepRecCluster obj's underlying member method code with example usage: obj.fit_ray_xshards(train_shards, val_shards) and returns: worker_stats, worker, worker