code: def _run_executor(args, pipeline_args) -> None:
      r"""Select a particular executor and run it based on name.
  
    # pylint: disable=line-too-long
    _run_executor() is used to invoke a class subclassing
    tfx.dsl.components.base.base_executor.BaseExecutor.  This function can be used
    for
    both invoking the executor on remote environments as well as for unit testing
    of executors.
  
    How to invoke an executor as standalone:
    # TODO(b/132958430): Create utility script to generate arguments for
    run_executor.py
    First, the input data needs to be prepared.  An easy way to generate the test
    data is to fully run the pipeline once.  This will generate the data to be
    used for testing as well as log the artifacts to be used as input parameters.
    In each executed component, three log entries will be generated similar to the
    below:
    ```
    [2019-05-16 08:59:27,117] {logging_mixin.py:95} INFO - [2019-05-16
    08:59:27,116] {base_executor.py:72} INFO - Starting Executor execution.
    [2019-05-16 08:59:27,117] {logging_mixin.py:95} INFO - [2019-05-16
    08:59:27,117] {base_executor.py:74} INFO - Inputs for Executor is:
    {"input_base": [{"artifact": {"id": "1", "typeId": "1", "uri":
    "/usr/local/google/home/khaas/taxi/data/simple", "properties": {"split":
    {"stringValue": ""}, "state": {"stringValue": "published"}, "span":
    {"intValue": "1"}, "type_name": {"stringValue": "ExternalPath"}}},
    "artifact_type": {"id": "1", "name": "ExternalPath", "properties": {"span":
    "INT", "name": "STRING", "type_name": "STRING", "split": "STRING", "state":
    "STRING"}}}]}
    [2019-05-16 08:59:27,117] {logging_mixin.py:95} INFO - [2019-05-16
    08:59:27,117] {base_executor.py:76} INFO - Outputs for Executor is:
    {"examples": [{"artifact": {"uri":
    "/usr/local/google/home/khaas/tfx/pipelines/chicago_taxi_simple/CsvExampleGen/examples/1/train/",
    "properties": {"type_name": {"stringValue": "ExamplesPath"}, "split":
    {"stringValue": "train"}, "span": {"intValue": "1"}}}, "artifact_type":
    {"name": "ExamplesPath", "properties": {"name": "STRING", "type_name":
    "STRING", "split": "STRING", "state": "STRING", "span": "INT"}}}, {"artifact":
    {"uri":
    "/usr/local/google/home/khaas/tfx/pipelines/chicago_taxi_simple/CsvExampleGen/examples/1/eval/",
    "properties": {"type_name": {"stringValue": "ExamplesPath"}, "split":
    {"stringValue": "eval"}, "span": {"intValue": "1"}}}, "artifact_type":
    {"name": "ExamplesPath", "properties": {"name": "STRING", "type_name":
    "STRING", "split": "STRING", "state": "STRING", "span": "INT"}}}]}
    [2019-05-16 08:59:27,117] {logging_mixin.py:95} INFO - [2019-05-16
    08:59:27,117] {base_executor.py:78} INFO - Execution properties for Executor
    is: {"output": "{  \"splitConfig\": {\"splits\": [{\"name\": \"train\",
    \"hashBuckets\": 2}, {\"name\": \"eval\",\"hashBuckets\": 1}]}}"}
    ```
    Each of these map directly to the input parameters expected by run_executor():
    ```
    python scripts/run_executor.py \
        --executor_class_path=tfx.components.example_gen.csv_example_gen.executor.Executor
        \
        --inputs={"input_base": [{"artifact": {"id": "1", "typeId": "1", "uri":
        "/usr/local/google/home/khaas/taxi/data/simple", "properties": {"split":
        {"stringValue": ""}, "state": {"stringValue": "published"}, "span":
        {"intValue": "1"}, "type_name": {"stringValue": "ExternalPath"}}},
        "artifact_type": {"id": "1", "name": "ExternalPath", "properties":
        {"span": "INT", "name": "STRING", "type_name": "STRING", "split":
        "STRING", "state": "STRING"}}}]} \
        --outputs={"examples": [{"artifact": {"uri":
        "/usr/local/google/home/khaas/tfx/pipelines/chicago_taxi_simple/CsvExampleGen/examples/1/train/",
        "properties": {"type_name": {"stringValue": "ExamplesPath"}, "split":
        {"stringValue": "train"}, "span": {"intValue": "1"}}}, "artifact_type":
        {"name": "ExamplesPath", "properties": {"name": "STRING", "type_name":
        "STRING", "split": "STRING", "state": "STRING", "span": "INT"}}},
        {"artifact": {"uri":
        "/usr/local/google/home/khaas/tfx/pipelines/chicago_taxi_simple/CsvExampleGen/examples/1/eval/",
        "properties": {"type_name": {"stringValue": "ExamplesPath"}, "split":
        {"stringValue": "eval"}, "span": {"intValue": "1"}}}, "artifact_type":
        {"name": "ExamplesPath", "properties": {"name": "STRING", "type_name":
        "STRING", "split": "STRING", "state": "STRING", "span": "INT"}}}]} \
        --exec-properties={"output": "{  \"splitConfig\": {\"splits\": [{\"name\":
        \"train\", \"hashBuckets\": 2}, {\"name\": \"eval\",\"hashBuckets\":
        1}]}}"}
    ```
    # pylint: disable=line-too-long
  
    Args:
      args:
        - inputs: The input artifacts for this execution, serialized as JSON.
        - outputs: The output artifacts to be generated by this execution,
          serialized as JSON.
        - exec_properties: The execution properties to be used by this execution,
          serialized as JSON.
      pipeline_args: Optional parameter that maps to the optional_pipeline_args
        parameter in the pipeline, which provides additional configuration options
        for apache-beam and tensorflow.logging.
  
    Returns:
      None
  
    Raises:
      None
    """
  
      absl.logging.set_verbosity(absl.logging.INFO)
  
      (inputs_str, outputs_str, exec_properties_str) = (
          args.inputs or base64.b64decode(args.inputs_base64),
          args.outputs or base64.b64decode(args.outputs_base64),
          args.exec_properties
          or base64.b64decode(args.exec_properties_base64),
      )
  
      inputs = artifact_utils.parse_artifact_dict(inputs_str)
      outputs = artifact_utils.parse_artifact_dict(outputs_str)
      exec_properties = json.loads(exec_properties_str)
      absl.logging.info(
          "Executor {} do: inputs: {}, outputs: {}, exec_properties: {}".format(
              args.executor_class_path, inputs, outputs, exec_properties
          )
      )
      executor_cls = import_utils.import_class_by_path(
          args.executor_class_path
      )
      if issubclass(executor_cls, base_beam_executor.BaseBeamExecutor):
          executor_context = (
              base_beam_executor.BaseBeamExecutor.Context(
                  beam_pipeline_args=pipeline_args,
                  tmp_dir=args.temp_directory_path,
                  unique_id="",
              )
          )
      else:
          executor_context = base_executor.BaseExecutor.Context(
              extra_flags=pipeline_args,
              tmp_dir=args.temp_directory_path,
              unique_id="",
          )
      executor = executor_cls(executor_context)
      absl.logging.info("Starting executor")
      executor.Do(inputs, outputs, exec_properties)
  
      # The last line of stdout will be pushed to xcom by Airflow.
      if args.write_outputs_stdout:
          print(artifact_utils.jsonify_artifact_dict(outputs))
  
file_path: tfx/tfx/scripts/run_executor.py
imports: import argparse
  import base64
  import json
  from typing import List, Tuple
  import absl
  from absl import app
  from absl.flags import argparse_flags
  from tfx.dsl.components.base import base_beam_executor
  from tfx.dsl.components.base import base_executor
  from tfx.types import artifact_utils
  from tfx.utils import import_utils
label_desc: Use image node_id 1 for calling a global function with example usage: _run_executor(args, pipeline_args) without return types