class_docstr: DoFn that executes one subgraph layer.
class_docstr_tok: ["DoFn","that","executes","one","subgraph","layer","."]
docstr: Executes a subgraph layer.
  
  To execute a subgraph layer, we need to prepare a feed_dict by extracting
  tensor values from element. Then, we run the subgraph and store its outputs
  to a copy of element.
  
  Since we import `GraphDef` protos, all the node names now have the prefix
  "import/". Also, TensorFlow feed_dict and outputs accept tensor
  names instead of node names. Hence, a conversion from node_name to
  "import/node_name:0" is necessary. Note that this conversion assumes
  that there is one output per node.
  
  Args:
    element: A dictionary from remote op names to a dictionary from tensor
      names to values. Element[remote_op_name] stores graph inputs and
      previous specs' outputs.
    spec: An ExecutionSpec for a subgraph layer.
    remote_op_name: The remote op name of the current graph.
  
  Yields:
    A dictionary from remote op names to a dictionary from tensor names to
    values. The dictionary is a copy of the input element, to which the
    outputs of this subgraph layer have been added.
docstr_tok: ["Executes","a","subgraph","layer",".","To","execute","a","subgraph","layer",",","we","need","to","prepare","a","feed_dict","by","extracting","tensor","values","from","element",".","Then",",","we","run","the","subgraph","and","store","its","outputs","to","a","copy","of","element",".","Since","we","import","`","GraphDef","`","protos",",","all","the","node","names","now","have","the","prefix","''","import\/","''",".","Also",",","TensorFlow","feed_dict","and","outputs","accept","tensor","names","instead","of","node","names",".","Hence",",","a","conversion","from","node_name","to","''","import\/node_name:0","''","is","necessary",".","Note","that","this","conversion","assumes","that","there","is","one","output","per","node",".","Args",":","element",":","A","dictionary","from","remote","op","names","to","a","dictionary","from","tensor","names","to","values",".","Element","[","remote_op_name","]","stores","graph","inputs","and","previous","specs","'","outputs",".","spec",":","An","ExecutionSpec","for","a","subgraph","layer",".","remote_op_name",":","The","remote","op","name","of","the","current","graph",".","Yields",":","A","dictionary","from","remote","op","names","to","a","dictionary","from","tensor","names","to","values",".","The","dictionary","is","a","copy","of","the","input","element",",","to","which","the","outputs","of","this","subgraph","layer","have","been","added","."]
code: def process(
      self,
      # Not using mapping here because it doesn't support item assignment.
      element: Dict[str, Dict[str, Any]],
      spec: execution_spec.ExecutionSpec,
      remote_op_name: str,
  ) -> Iterator[Dict[str, Dict[str, Any]]]:
      """Executes a subgraph layer.
  
      To execute a subgraph layer, we need to prepare a feed_dict by extracting
      tensor values from element. Then, we run the subgraph and store its outputs
      to a copy of element.
  
      Since we import `GraphDef` protos, all the node names now have the prefix
      "import/". Also, TensorFlow feed_dict and outputs accept tensor
      names instead of node names. Hence, a conversion from node_name to
      "import/node_name:0" is necessary. Note that this conversion assumes
      that there is one output per node.
  
      Args:
        element: A dictionary from remote op names to a dictionary from tensor
          names to values. Element[remote_op_name] stores graph inputs and
          previous specs' outputs.
        spec: An ExecutionSpec for a subgraph layer.
        remote_op_name: The remote op name of the current graph.
  
      Yields:
        A dictionary from remote op names to a dictionary from tensor names to
        values. The dictionary is a copy of the input element, to which the
        outputs of this subgraph layer have been added.
      """
      element = copy.deepcopy(element)
      input_tensor_names = [
          _import_tensor_name(node_name)
          for node_name in spec.input_names
      ]
      output_tensor_names = [
          _import_tensor_name(node_name)
          for node_name in spec.output_names
      ]
      feed_dict = {
          tensor_name: element[remote_op_name][tensor_name]
          for tensor_name in input_tensor_names
      }
  
      outputs = []
      with tf.compat.v1.Session(graph=tf.Graph()) as sess:
          tf.import_graph_def(spec.subgraph)
          outputs = sess.run(output_tensor_names, feed_dict=feed_dict)
  
      for output_tensor_name, output_tensor in zip(
          output_tensor_names, outputs
      ):
          element[remote_op_name][output_tensor_name] = output_tensor
  
      yield element
  
file_path: tfx/tfx/experimental/distributed_inference/graphdef_experiments/subgraph_partitioning/beam_pipeline.py
imports: import copy
  from typing import Any, Dict, Iterator, List, Mapping
  import apache_beam
  import tensorflow
  from tfx.experimental.distributed_inference.graphdef_experiments.subgraph_partitioning import execution_spec
label_desc: Use image node_id 1 for calling the _SubgraphLayerDoFn obj's underlying member method code with example usage: obj.process(element, spec, remote_op_name) without return types