code: def word_count(
      input_path, output_path, raw_metadata, min_token_frequency=2
  ):
      """Returns a pipeline counting words and writing the output.
  
      Args:
        input_path: recordio file to read
        output_path: path in which to write the output
        raw_metadata: metadata of input tf.Examples
        min_token_frequency: the min frequency for a token to be included
      """
  
      lang_set = set(FLAGS.lang_set.split(","))
  
      # Create pipeline.
      pipeline = beam.Pipeline()
  
      with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
          converter = tft.coders.ExampleProtoCoder(
              raw_metadata.schema, serialized=False
          )
  
          # Read raw data and convert to TF Transform encoded dict.
          raw_data = (
              pipeline
              | "ReadInputData"
              >> beam.io.tfrecordio.ReadFromTFRecord(
                  input_path,
                  coder=beam.coders.ProtoCoder(tf.train.Example),
              )
              | "DecodeInputData" >> beam.Map(converter.decode)
          )
  
          # Apply TF Transform.
          (transformed_data, _), _ = (
              raw_data,
              raw_metadata,
          ) | "FilterLangAndExtractToken" >> tft_beam.AnalyzeAndTransformDataset(
              utils.count_preprocessing_fn(
                  FLAGS.text_key, FLAGS.language_code_key
              )
          )
  
          # Filter by languages.
          tokens = transformed_data | "FilterByLang" >> beam.ParDo(
              utils.FilterTokensByLang(lang_set)
          )
  
          # Calculate smoothing coefficients.
          coeffs = (
              tokens
              | "CalculateSmoothingCoefficients"
              >> beam.CombineGlobally(
                  utils.CalculateCoefficients(FLAGS.smoothing_exponent)
              )
          )
  
          # Apply smoothing, aggregate counts, and sort words by count.
          _ = (
              tokens
              | "ApplyExponentialSmoothing"
              >> beam.ParDo(
                  utils.ExponentialSmoothing(),
                  beam.pvalue.AsSingleton(coeffs),
              )
              | "SumCounts" >> beam.CombinePerKey(sum)
              | "FilterLowCounts"
              >> beam.ParDo(
                  utils.FilterByCount(
                      FLAGS.max_word_length, min_token_frequency
                  )
              )
              | "MergeAndSortCounts"
              >> beam.CombineGlobally(utils.SortByCount())
              | "Flatten" >> beam.FlatMap(lambda x: x)
              | "FormatCounts"
              >> beam.Map(lambda tc: "%s\t%s" % (tc[0], tc[1]))
              | "WriteSortedCount"
              >> beam.io.WriteToText(
                  output_path, shard_name_template=""
              )
          )
  
      return pipeline
  
file_path: text/tensorflow_text/tools/wordpiece_vocab/generate_word_counts.py
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import tempfile
  from absl import app
  from absl import flags
  import apache_beam
  import tensorflow.compat.v1
  import tensorflow_transform
  import tensorflow_transform.beam
  from tensorflow_transform.tf_metadata import dataset_metadata
  from tensorflow_transform.tf_metadata import schema_utils
  from tensorflow_text.tools.wordpiece_vocab import utils
label_desc: Use image node_id 1 for calling a global function with example usage: word_count(input_path, output_path, raw_metadata, min_token_frequency) and returns: pipeline