code: def calculate_metrics():
      """Returns a pipeline to compute wordpiece model stats given a vocab and corpus."""
  
      # Schema of input dataset.
      raw_metadata = dataset_metadata.DatasetMetadata(
          schema_utils.schema_from_feature_spec(
              {
                  "text": tf.FixedLenFeature([], tf.string),
                  "language_code": tf.FixedLenFeature([], tf.string),
              }
          )
      )
  
      # Schema to format metrics as CSV.
      csv_schema = schema_utils.schema_from_feature_spec(
          {
              "lang": tf.FixedLenFeature([], tf.string),
              "sample_count": tf.FixedLenFeature([], tf.int64),
              "micro_drop_char_percent": tf.FixedLenFeature(
                  [], tf.string
              ),
              "macro_drop_char_percent": tf.FixedLenFeature(
                  [], tf.string
              ),
              "micro_compress_ratio": tf.FixedLenFeature([], tf.string),
              "macro_compress_ratio": tf.FixedLenFeature([], tf.string),
              "unweighted_en_wp_overlap_percent": tf.FixedLenFeature(
                  [], tf.string
              ),
              "weighted_en_wp_overlap_percent": tf.FixedLenFeature(
                  [], tf.string
              ),
          }
      )
  
      columns = [
          "lang",
          "sample_count",
          "micro_drop_char_percent",
          "macro_drop_char_percent",
          "micro_compress_ratio",
          "macro_compress_ratio",
          "unweighted_en_wp_overlap_percent",
          "weighted_en_wp_overlap_percent",
      ]
  
      # Create pipeline.
      pipeline = beam.Pipeline()
  
      with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
          example_converter = tft.coders.ExampleProtoCoder(
              raw_metadata.schema, serialized=False
          )
          csv_converter = tft.coders.CsvCoder(columns, csv_schema)
  
          # Read raw data and convert to TF Transform encoded dict.
          raw_data = (
              pipeline
              | "ReadInputData"
              >> beam.io.tfrecordio.ReadFromTFRecord(
                  FLAGS.input_file,
                  coder=beam.coders.ProtoCoder(tf.train.Example),
              )
              | "DecodeInputData" >> beam.Map(example_converter.decode)
          )
  
          # Apply transform to wordpiece-tokenize input.
          (transformed_data, _), _ = (
              raw_data,
              raw_metadata,
          ) | "WordpieceTokenizeInput" >> tft_beam.AnalyzeAndTransformDataset(
              utils.metrics_preprocessing_fn(
                  FLAGS.vocab_file,
                  FLAGS.text_key,
                  FLAGS.language_code_key,
              )
          )
  
          # Aggregate values for each lang, calculate metrics, and write to output.
          _ = (
              transformed_data
              | "CompileTokenInfo"
              >> beam.ParDo(utils.CompileTokenizationInfo())
              | "CombineStatsForLang"
              >> beam.CombineGlobally(utils.AggregateLang())
              | "CalculateMetrics"
              >> beam.ParDo(utils.CalculateMetrics())
              | "EncodeMetrics" >> beam.Map(csv_converter.encode)
              | "WriteMetrics"
              >> beam.io.WriteToText(
                  FLAGS.output_file,
                  shard_name_template="",
                  header=",".join(columns),
              )
          )
  
      return pipeline
  
file_path: text/tensorflow_text/tools/wordpiece_vocab/measure_wordpiece_stats.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: calculate_metrics() and returns: pipeline