code: def testVocabularyWithUserDefinedLookupFnFeedsSecondAnalyzer(self):
      input_data = [
          dict(x=b"bar"),
          dict(x=b"foo"),
          dict(x=b"bar"),
          dict(x=b"bar"),
          dict(x=b"foo"),
      ]
      input_metadata = tft.DatasetMetadata.from_feature_spec(
          {"x": tf.io.FixedLenFeature([], tf.string)}
      )
      expected_data = [
          dict(x=b"bar", x_int=0, x_int_mean=0.4),
          dict(x=b"bar", x_int=0, x_int_mean=0.4),
          dict(x=b"bar", x_int=0, x_int_mean=0.4),
          dict(x=b"foo", x_int=1, x_int_mean=0.4),
          dict(x=b"foo", x_int=1, x_int_mean=0.4),
      ]
      expected_vocab_file_contents = [(b"bar"), (b"foo")]
      size = len(expected_vocab_file_contents) - 1
      expected_metadata = tft.DatasetMetadata.from_feature_spec(
          {
              "x": tf.io.FixedLenFeature([], tf.string),
              "x_int": tf.io.FixedLenFeature([], tf.int64),
              "x_int_mean": tf.io.FixedLenFeature([], tf.float32),
          },
          domains={
              "x_int": schema_pb2.IntDomain(
                  min=-1, max=size, is_categorical=True
              )
          },
      )
  
      def preprocessing_fn(inputs):
          def _make_table_initializer(filename_tensor):
              if self._VocabFormat() == "text":
                  return tf.lookup.TextFileInitializer(
                      filename=filename_tensor,
                      key_dtype=tf.string,
                      key_index=tf.lookup.TextFileIndex.WHOLE_LINE,
                      value_dtype=tf.int64,
                      value_index=tf.lookup.TextFileIndex.LINE_NUMBER,
                  )
              elif self._VocabFormat() == "tfrecord_gzip":
                  return tft.tf_utils.make_tfrecord_vocabulary_lookup_initializer(
                      filename_tensor, return_indicator_as_value=False
                  )
  
          def _apply_vocab(y, deferred_vocab_filename_tensor):
              initializer = _make_table_initializer(
                  deferred_vocab_filename_tensor
              )
              table = tf.lookup.StaticHashTable(
                  initializer, default_value=-1
              )
              table_size = table.size()
              return table.lookup(y), table_size
  
          deferred_vocab_and_filename = tft.vocabulary(
              inputs["x"],
              vocab_filename="my_vocab",
              file_format=self._VocabFormat(),
          )
          x_int = tft.apply_vocabulary(
              inputs["x"],
              deferred_vocab_and_filename,
              lookup_fn=_apply_vocab,
              file_format=self._VocabFormat(),
          )
  
          x_int_mean = tf.zeros_like(
              x_int, dtype=tf.float32
          ) + tft.mean(x_int)
          return {
              "x": inputs["x"],
              "x_int": x_int,
              "x_int_mean": x_int_mean,
          }
  
      self.assertAnalyzeAndTransformResults(
          input_data,
          input_metadata,
          preprocessing_fn,
          expected_data,
          expected_metadata,
          expected_vocab_file_contents={
              "my_vocab": expected_vocab_file_contents
          },
      )
  
imports: import os
  import apache_beam
  import tensorflow
  import tensorflow_transform
  from tensorflow_transform.beam import analyzer_impls
  from tensorflow_transform.beam import impl
  from tensorflow_transform.beam.tft_beam_io import transform_fn_io
  from tensorflow_transform.beam import tft_unit
  from tensorflow_metadata.proto.v0 import schema_pb2
label_desc: Use image node_id 16 for calling the VocabularyIntegrationTest obj's underlying member method code with example usage: obj.testVocabularyWithUserDefinedLookupFnFeedsSecondAnalyzer() and returns: dict, table,  table_size, tf, tft