class_docstr: Writes a TransformFn to disk.
  
  The internal structure is a directory containing two subdirectories.  The
  first is 'transformed_metadata' and contains metadata of the transformed data.
  The second is 'transform_fn' and contains a SavedModel representing the
  transformed data.
class_docstr_tok: ["Writes","a","TransformFn","to","disk",".","The","internal","structure","is","a","directory","containing","two","subdirectories",".","The","first","is","'transformed_metadata","'","and","contains","metadata","of","the","transformed","data",".","The","second","is","'transform_fn","'","and","contains","a","SavedModel","representing","the","transformed","data","."]
code: def expand(self, transform_fn):
      saved_model_dir, metadata = transform_fn
      pipeline = saved_model_dir.pipeline
  
      # Using a temp dir within `path` ensures that the source and dstination
      # paths for the rename below are in the same file system.
      base_temp_dir = os.path.join(self._path, "transform_tmp")
      temp_metadata_path = (
          metadata
          | "WriteMetadataToTemp"
          >> beam_metadata_io.WriteMetadata(
              base_temp_dir, pipeline, write_to_unique_subdirectory=True
          )
      )
  
      temp_transform_fn_path = (
          saved_model_dir
          | "WriteTransformFnToTemp"
          >> beam.Map(_copy_tree_to_unique_temp_dir, base_temp_dir)
      )
  
      metadata_path = os.path.join(
          self._path, tft.TFTransformOutput.TRANSFORMED_METADATA_DIR
      )
      transform_fn_path = os.path.join(
          self._path, tft.TFTransformOutput.TRANSFORM_FN_DIR
      )
  
      def publish_outputs(
          unused_element, metadata_source_path, transform_fn_source_path
      ):
          import tensorflow as tf  # pylint: disable=g-import-not-at-top
  
          if not tf.io.gfile.exists(self._path):
              tf.io.gfile.makedirs(self._path)
  
          tf.io.gfile.rename(
              metadata_source_path, metadata_path, overwrite=True
          )
          tf.io.gfile.rename(
              transform_fn_source_path,
              transform_fn_path,
              overwrite=True,
          )
          # TODO(b/211615643): Remove the exists check once importing TFIO in S3
          # addresses NotFoundError.
          if tf.io.gfile.exists(base_temp_dir):
              tf.io.gfile.rmtree(base_temp_dir)
  
      # TODO(KesterTong): Move this "must follows" logic into a tfx_bsl helper
      # function or into Beam.
      return (
          pipeline
          | "CreateSole" >> beam.Create([None])
          | "PublishMetadataAndTransformFn"
          >> beam.Map(
              publish_outputs,
              metadata_source_path=beam.pvalue.AsSingleton(
                  temp_metadata_path
              ),
              transform_fn_source_path=beam.pvalue.AsSingleton(
                  temp_transform_fn_path
              ),
          )
      )
  
file_path: transform/tensorflow_transform/beam/tft_beam_io/transform_fn_io.py
imports: import os
  import apache_beam
  import tensorflow_transform
  from tensorflow_transform import impl_helper
  from tensorflow_transform.beam import common
  from tensorflow_transform.beam.tft_beam_io import beam_metadata_io
  from tensorflow_transform.tf_metadata import metadata_io
label_desc: Use image node_id 3 for calling the WriteTransformFn obj's underlying member method code with example usage: obj.expand(transform_fn) and returns: unknown