class_docstr: Represents a distributed set of elements backed by an RDD,
  which is created by applying tensorflow dataset transformations
  on each partitions.
code: def as_graph_rdd(
      self, batch_per_shard: int, drop_remainder: bool = True
  ) -> "RDD[Any]":
      create_dataset_fn = self.create_dataset_fn
  
      def to_dataset(iter):
          data_list = list(iter)
  
          import tensorflow as tf
  
          if not data_list:
              return []
  
          datasets = [create_dataset_fn(data) for data in data_list]
          from functools import reduce
  
          dataset = reduce(lambda x, y: x.concatenate(y), datasets)
          dataset = dataset.batch(batch_per_shard, drop_remainder)
          iterator = dataset.make_initializable_iterator()
          train_next_ops = nest.flatten(iterator.get_next())
          output_types = [t for t in nest.flatten(dataset.output_types)]
          output_types_enum = [t.as_datatype_enum for t in output_types]
  
          init_op_name = iterator.initializer.name
          table_init_op = tf.tables_initializer().name
          output_names = [op.name for op in train_next_ops]
  
          graph = train_next_ops[0].graph
  
          flatten_shapes = nest.flatten(dataset.output_shapes)
  
          flatten_shapes = [shape[1:] for shape in flatten_shapes]
  
          flatten_tensor_structure = [
              TensorMeta(
                  dtype=output_types[i],
                  shape=list(flatten_shapes[i]),
                  name="zoo_input_{}".format(i),
              )
              for i in range(len(flatten_shapes))
          ]
          structure = dataset.output_types
          if isinstance(structure, tf.DType):
              structure = (structure,)
          tensor_structure = nest.pack_sequence_as(
              structure, flatten_tensor_structure
          )
  
          meta_info = {
              "init_op_name": init_op_name,
              "table_init_op": table_init_op,
              "output_names": output_names,
              "output_types": output_types_enum,
              "tensor_structure": tensor_structure,
          }
  
          return [
              (
                  bytearray(graph.as_graph_def().SerializeToString()),
                  meta_info,
              )
          ]
  
      graph_rdd_and_meta = self.xshards.rdd.mapPartitions(to_dataset)
      return graph_rdd_and_meta
  
imports: import tensorflow
  from bigdl.orca.tfpark.tf_dataset import TensorMeta
  from bigdl.dllib.utils import nest
  from bigdl.orca.data import SparkXShards
  from bigdl.dllib.utils import log4Error
  from bigdl.dllib.utils.log4Error import invalidInputError
  from typing import TYPE_CHECKING, Callable, Optional, Any
label_desc: Use image node_id 2 for calling the Dataset obj's underlying member method code with example usage: obj.as_graph_rdd(batch_per_shard, drop_remainder) and returns: graph_rdd_and_meta, list, list