code: def xshards_to_tf_dataset(
      data_shard,
      batch_size=-1,
      batch_per_thread=-1,
      validation_data_shard=None,
      hard_code_batch_size=False,
      memory_type="DRAM",
      sequential_order=False,
      shuffle=True,
  ):
      # todo data_shard.head ?
      feature_spec, label_spec = data_shard._for_each(
          get_spec(allow_tuple=True, allow_list=False)
      ).first()
  
      feature_spec = [
          (tf.dtypes.as_dtype(spec[0]), spec[1])
          for spec in feature_spec
      ]
      label_spec = (
          [
              (tf.dtypes.as_dtype(spec[0]), spec[1])
              for spec in label_spec
          ]
          if label_spec is not None
          else None
      )
  
      invalidInputError(
          batch_size != -1 or batch_per_thread != -1,
          "one of batch_size and batch_per_thread should be specified",
      )
  
      val_rdd = (
          None
          if validation_data_shard is None
          else validation_data_shard.rdd.flatMap(
              flatten_xy(allow_tuple=True, allow_list=False)
          )
      )
  
      dataset = TFDataset.from_rdd(
          data_shard.rdd.flatMap(
              flatten_xy(allow_tuple=True, allow_list=False)
          ),
          features=feature_spec,
          labels=label_spec,
          batch_size=batch_size,
          batch_per_thread=batch_per_thread,
          val_rdd=val_rdd,
          hard_code_batch_size=hard_code_batch_size,
          memory_type=memory_type,
          sequential_order=sequential_order,
          shuffle=shuffle,
      )
  
      return dataset
  
imports: import tensorflow
  from bigdl.orca.tfpark.tf_dataset import TFDataset
  from bigdl.orca.data.utils import get_spec, flatten_xy
  from bigdl.dllib.utils.log4Error import *
label_desc: Use image node_id 1 for calling a global function with example usage: xshards_to_tf_dataset(data_shard, batch_size, batch_per_thread, validation_data_shard, hard_code_batch_size, memory_type, sequential_order, shuffle) and returns: dataset