class_docstr: TFX Tuner executor impl as a worker in a Google Cloud AI Platform job.
class_docstr_tok: ["TFX","Tuner","executor","impl","as","a","worker","in","a","Google","Cloud","AI","Platform","job","."]
code: def __init__(self, context):
      super().__init__(context)
  
      # Those fields are populated only when running in distribution.
      self._is_chief = False
      self._tuner_id = None
      self._master_addr = None
      self._master_port = None
  
      # Initialize configuration of distribution according to CLUSTER_SPEC
      logging.info("Initializing cluster spec... ")
  
      cluster_spec = json.loads(os.environ.get("CLUSTER_SPEC", "{}"))
  
      # If CLUSTER_SPEC is not present, assume single-machine tuning.
      if not cluster_spec:
          return
  
      logging.info("Cluster spec initalized with: %s", cluster_spec)
  
      # task_type can be 'master', 'worker', 'chief' or the type of worker pool.
      # 'workerpool0' refers to the primary replica.
      task_type = cluster_spec["task"]["type"]
      if "workerpool0" in cluster_spec["cluster"]:
          self._master_addr, self._master_port = (
              # We rely on Vertex AI Training service's specification whereby
              # there will be no more than one primary replica.
              # https://cloud.google.com/vertex-ai/docs/training/distributed-training#cluster-spec-format
              cluster_spec["cluster"]["workerpool0"][0].split(":")
          )
          self._is_chief = task_type == "workerpool0"
      elif task_type == "chief" or task_type == "workerpool0":
          self._master_addr, self._master_port = (
              # CLUSTER_SPEC is different when only primary replica is present
              # in Vertex AI Training.
              cluster_spec["cluster"]["chief"][0].split(":")
          )
          self._is_chief = True
          task_type = "workerpool0"
      else:
          self._master_addr, self._master_port = (
              # We rely on Cloud AI Platform Training service's specification
              # whereby there will be no more than one master replica.
              # https://cloud.google.com/ai-platform/training/docs/distributed-training-containers#cluster-spec-format
              cluster_spec["cluster"]["master"][0].split(":")
          )
          self._is_chief = task_type == "master"
  
      self._tuner_id = "tfx-tuner-%s-%d" % (
          task_type,
          cluster_spec["task"]["index"],  # zero-based index
      )
  
      logging.info("Tuner ID is: %s", self._tuner_id)
  
imports: import datetime
  import json
  import os
  from typing import Any, Dict, List
  from absl import logging
  from tfx import types
  from tfx.components.tuner import executor
  from tfx.dsl.components.base import base_executor
  from tfx.extensions.google_cloud_ai_platform import constants
  from tfx.extensions.google_cloud_ai_platform import runner
  from tfx.extensions.google_cloud_ai_platform.trainer import executor
  from tfx.types import standard_component_specs
  from tfx.utils import doc_controls
  from tfx.utils import json_utils
  from tfx.utils import name_utils
  import multiprocessing
label_desc: Use image node_id 4 to create a new _WorkerExecutor object from inherited base classes: base_executor with example: obj = _WorkerExecutor(context)