class_docstr: AI Platform Training executor.
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      self._log_startup(input_dict, output_dict, exec_properties)
  
      aip_config = json_utils.loads(exec_properties.pop(CONFIG_KEY))
  
      assert aip_config, "AIP training config is not found."
  
      training_job = aip_config.pop(TRAINING_JOB_CONFIG_KEY)
      job_id = aip_config.pop(JOB_ID_CONFIG_KEY)
      project = aip_config.pop(PROJECT_CONFIG_KEY)
  
      # Resolve parameters.
      training_job["training_input"][
          "args"
      ] = container_common._resolve_container_command_line(  # pylint: disable=protected-access
          cmd_args=training_job["training_input"]["args"],
          input_dict=input_dict,
          output_dict=output_dict,
          exec_properties=exec_properties,
      )
  
      training_job["job_id"] = job_id or "tfx_{}".format(
          datetime.datetime.now().strftime("%Y%m%d%H%M%S")
      )
      # Invoke CMLE job
      runner._launch_cloud_training(  # pylint: disable=protected-access
          project=project, training_job=training_job
      )
  
file_path: tfx/tfx/orchestration/kubeflow/v2/components/experimental/ai_platform_training_executor.py
imports: import datetime
  from typing import Any, Dict, List
  from tfx import types
  from tfx.dsl.components.base import base_executor
  from tfx.extensions.google_cloud_ai_platform import runner
  from tfx.orchestration.launcher import container_common
  from tfx.utils import json_utils
label_desc: Use image node_id 1 for calling the AiPlatformTrainingExecutor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types