code: def create(self):
      # A discerning eye would notice and question the choice of using the
      # V1Job construct over the V1Pod construct given that we don't rely much
      # on any of the V1Job semantics. The major reasons at the moment are -
      #     1. It makes the Kubernetes UIs (Octant, Lens) a bit easier on
      #        the eyes, although even that can be questioned.
      #     2. AWS Step Functions, at the moment (Apr' 22) only supports
      #        executing Jobs and not Pods as part of it's publicly declared
      #        API. When we ship the AWS Step Functions integration with EKS,
      #        it will hopefully lessen our workload.
      #
      # Note: This implementation ensures that there is only one unique Pod
      # (unique UID) per Metaflow task attempt.
      client = self._client.get()
  
      # tmpfs variables
      use_tmpfs = self._kwargs["use_tmpfs"]
      tmpfs_size = self._kwargs["tmpfs_size"]
      tmpfs_enabled = use_tmpfs or (tmpfs_size and not use_tmpfs)
  
      self._job = client.V1Job(
          api_version="batch/v1",
          kind="Job",
          metadata=client.V1ObjectMeta(
              # Annotations are for humans
              annotations=self._kwargs.get("annotations", {}),
              # While labels are for Kubernetes
              labels=self._kwargs.get("labels", {}),
              generate_name=self._kwargs["generate_name"],
              namespace=self._kwargs[
                  "namespace"
              ],  # Defaults to `default`
          ),
          spec=client.V1JobSpec(
              # Retries are handled by Metaflow when it is responsible for
              # executing the flow. The responsibility is moved to Kubernetes
              # when Argo Workflows is responsible for the execution.
              backoff_limit=self._kwargs.get("retries", 0),
              completions=1,  # A single non-indexed pod job
              ttl_seconds_after_finished=7
              * 60
              * 60  # Remove job after a week. TODO: Make this configurable
              * 24,
              template=client.V1PodTemplateSpec(
                  metadata=client.V1ObjectMeta(
                      annotations=self._kwargs.get("annotations", {}),
                      labels=self._kwargs.get("labels", {}),
                      namespace=self._kwargs["namespace"],
                  ),
                  spec=client.V1PodSpec(
                      # Timeout is set on the pod and not the job (important!)
                      active_deadline_seconds=self._kwargs[
                          "timeout_in_seconds"
                      ],
                      # TODO (savin): Enable affinities for GPU scheduling.
                      # affinity=?,
                      containers=[
                          client.V1Container(
                              command=self._kwargs["command"],
                              env=[
                                  client.V1EnvVar(name=k, value=str(v))
                                  for k, v in self._kwargs.get(
                                      "environment_variables", {}
                                  ).items()
                              ]
                              # And some downward API magic. Add (key, value)
                              # pairs below to make pod metadata available
                              # within Kubernetes container.
                              + [
                                  client.V1EnvVar(
                                      name=k,
                                      value_from=client.V1EnvVarSource(
                                          field_ref=client.V1ObjectFieldSelector(
                                              field_path=str(v)
                                          )
                                      ),
                                  )
                                  for k, v in {
                                      "METAFLOW_KUBERNETES_POD_NAMESPACE": "metadata.namespace",
                                      "METAFLOW_KUBERNETES_POD_NAME": "metadata.name",
                                      "METAFLOW_KUBERNETES_POD_ID": "metadata.uid",
                                      "METAFLOW_KUBERNETES_SERVICE_ACCOUNT_NAME": "spec.serviceAccountName",
                                      "METAFLOW_KUBERNETES_NODE_IP": "status.hostIP",
                                  }.items()
                              ]
                              + [
                                  client.V1EnvVar(name=k, value=str(v))
                                  for k, v in inject_tracing_vars(
                                      {}
                                  ).items()
                              ],
                              env_from=[
                                  client.V1EnvFromSource(
                                      secret_ref=client.V1SecretEnvSource(
                                          name=str(k),
                                          # optional=True
                                      )
                                  )
                                  for k in list(
                                      self._kwargs.get("secrets", [])
                                  )
                                  + KUBERNETES_SECRETS.split(",")
                                  if k
                              ],
                              image=self._kwargs["image"],
                              image_pull_policy=self._kwargs[
                                  "image_pull_policy"
                              ],
                              name=self._kwargs["step_name"].replace(
                                  "_", "-"
                              ),
                              resources=client.V1ResourceRequirements(
                                  requests={
                                      "cpu": str(self._kwargs["cpu"]),
                                      "memory": "%sM"
                                      % str(self._kwargs["memory"]),
                                      "ephemeral-storage": "%sM"
                                      % str(self._kwargs["disk"]),
                                  },
                                  limits={
                                      "%s.com/gpu".lower()
                                      % self._kwargs["gpu_vendor"]: str(
                                          self._kwargs["gpu"]
                                      )
                                      for k in [0]
                                      # Don't set GPU limits if gpu isn't specified.
                                      if self._kwargs["gpu"] is not None
                                  },
                              ),
                              volume_mounts=(
                                  [
                                      client.V1VolumeMount(
                                          mount_path=self._kwargs.get(
                                              "tmpfs_path"
                                          ),
                                          name="tmpfs-ephemeral-volume",
                                      )
                                  ]
                                  if tmpfs_enabled
                                  else []
                              )
                              + (
                                  [
                                      client.V1VolumeMount(
                                          mount_path=path, name=claim
                                      )
                                      for claim, path in self._kwargs[
                                          "persistent_volume_claims"
                                      ].items()
                                  ]
                                  if self._kwargs[
                                      "persistent_volume_claims"
                                  ]
                                  is not None
                                  else []
                              ),
                          )
                      ],
                      node_selector=self._kwargs.get("node_selector"),
                      # TODO (savin): Support image_pull_secrets
                      # image_pull_secrets=?,
                      # TODO (savin): Support preemption policies
                      # preemption_policy=?,
                      #
                      # A Container in a Pod may fail for a number of
                      # reasons, such as because the process in it exited
                      # with a non-zero exit code, or the Container was
                      # killed due to OOM etc. If this happens, fail the pod
                      # and let Metaflow handle the retries.
                      restart_policy="Never",
                      service_account_name=self._kwargs[
                          "service_account"
                      ],
                      # Terminate the container immediately on SIGTERM
                      termination_grace_period_seconds=0,
                      tolerations=[
                          client.V1Toleration(**toleration)
                          for toleration in self._kwargs.get(
                              "tolerations"
                          )
                          or []
                      ],
                      volumes=(
                          [
                              client.V1Volume(
                                  name="tmpfs-ephemeral-volume",
                                  empty_dir=client.V1EmptyDirVolumeSource(
                                      medium="Memory",
                                      # Add default unit as ours differs from Kubernetes default.
                                      size_limit="{}Mi".format(
                                          tmpfs_size
                                      ),
                                  ),
                              )
                          ]
                          if tmpfs_enabled
                          else []
                      )
                      + (
                          [
                              client.V1Volume(
                                  name=claim,
                                  persistent_volume_claim=client.V1PersistentVolumeClaimVolumeSource(
                                      claim_name=claim
                                  ),
                              )
                              for claim in self._kwargs[
                                  "persistent_volume_claims"
                              ].keys()
                          ]
                          if self._kwargs["persistent_volume_claims"]
                          is not None
                          else []
                      ),
                      # TODO (savin): Set termination_message_policy
                  ),
              ),
          ),
      )
      return self
  
imports: import json
  import math
  import random
  import time
  from metaflow.tracing import inject_tracing_vars
  from metaflow.exception import MetaflowException
  from metaflow.metaflow_config import KUBERNETES_SECRETS
label_desc: Use image node_id 2 for calling the KubernetesJob obj's underlying member method code with example usage: obj.create() and returns: self