code: def _run_transform(args, beam_pipeline_args):
      """Construct and run transform executor."""
      absl.logging.set_verbosity(absl.logging.INFO)
  
      def make_beam_pipeline():
          return beam.Pipeline(beam_pipeline_args)
  
      preprocessing_fn = udf_utils.get_fn(
          {
              standard_component_specs.PREPROCESSING_FN_KEY: args.preprocessing_fn_path
          },
          standard_component_specs.PREPROCESSING_FN_KEY,
      )
  
      inputs = {
          labels.ANALYZE_DATA_PATHS_LABEL: args.analyze_examples,
          labels.ANALYZE_PATHS_FILE_FORMATS_LABEL: [
              labels.FORMAT_TFRECORD
          ]
          * len(args.analyze_examples),
          labels.TRANSFORM_DATA_PATHS_LABEL: [
              args.analyze_examples + args.transform_only_examples
          ],
          labels.TRANSFORM_PATHS_FILE_FORMATS_LABEL: [
              labels.FORMAT_TFRECORD
          ]
          * (
              len(args.analyze_examples)
              + len(args.transform_only_examples)
          ),
          labels.SCHEMA_PATH_LABEL: args.input_schema_path,
          labels.PREPROCESSING_FN: preprocessing_fn,
          labels.EXAMPLES_DATA_FORMAT_LABEL: example_gen_pb2.PayloadFormat.Value(
              args.example_data_format
          ),
          labels.DISABLE_STATISTICS_LABEL: args.disable_statistics,
          labels.MAKE_BEAM_PIPELINE_FN: make_beam_pipeline,
      }
      outputs = {
          labels.TRANSFORM_METADATA_OUTPUT_PATH_LABEL: args.transform_fn,
          labels.TRANSFORM_MATERIALIZE_OUTPUT_PATHS_LABEL: (
              args.transformed_examples
          ),
          labels.PER_SET_STATS_OUTPUT_PATHS_LABEL: (
              args.per_set_stats_outputs
          ),
          labels.TEMP_OUTPUT_LABEL: args.tmp_location,
      }
  
      executor.TransformProcessor().Transform(
          inputs, outputs, args.status_file
      )
  
file_path: tfx/tfx/components/transform/run_executor.py
imports: import argparse
  from typing import List, Tuple
  import absl
  from absl import app
  from absl.flags import argparse_flags
  import apache_beam
  from tfx.components.transform import executor
  from tfx.components.transform import labels
  from tfx.components.util import udf_utils
  from tfx.proto import example_gen_pb2
  from tfx.types import standard_component_specs
label_desc: Use image node_id 1 for calling a global function with example usage: _run_transform(args, beam_pipeline_args) and returns: beam