KubernetesPodOperator ¶

KubernetesPodOperator 允许您在 Kubernetes 集群上创建和运行 Pod。

注意

如果您使用的是托管 Kubernetes，请考虑使用专门的 KPO 运算符，因为它简化了 Kubernetes 授权过程

用于 Google Kubernetes Engine 的 GKEStartPodOperator 运算符，
用于 AWS Elastic Kubernetes Engine 的 EksPodOperator 运算符。

注意

使用此运算符不需要 Kubernetes 执行器。

此运算符如何工作？¶

KubernetesPodOperator 使用 Kubernetes API 在 Kubernetes 集群中启动 Pod。通过提供镜像 URL 和带有可选参数的命令，该运算符使用 Kube Python 客户端生成 Kubernetes API 请求，从而动态启动这些单独的 Pod。用户可以使用 config_file 参数指定 kubeconfig 文件，否则运算符将默认为 ~/.kube/config。

KubernetesPodOperator 支持任务级资源配置，并且对于公共 PyPI 存储库中不可用的自定义 Python 依赖项来说是最佳选择。它还允许用户使用 pod_template_file 参数提供模板 YAML 文件。最终，它允许 Airflow 充当作业编排器 - 无论这些作业是用什么语言编写的。

调试 KubernetesPodOperator ¶

您可以通过在运算符实例上调用 dry_run() 来打印出将在运行时创建的 Pod 的 Kubernetes 清单。

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

k = KubernetesPodOperator(
    name="hello-dry-run",
    image="debian",
    cmds=["bash", "-cx"],
    arguments=["echo", "10"],
    labels={"foo": "bar"},
    task_id="dry_run_demo",
    do_xcom_push=True,
)

k.dry_run()

参数优先级 ¶

当 KPO 定义 Pod 对象时，KubernetesPodOperator 参数之间可能存在重叠。通常，优先级顺序是 KPO 字段特定参数（例如，secrets、cmds、affinity），更通用的模板 full_pod_spec、pod_template_file、pod_template_dict，最后是默认的 V1Pod。

对于 namespace，如果未通过任何这些方法提供命名空间，那么我们将首先尝试获取当前命名空间（如果任务已经在 Kubernetes 中运行），如果失败，我们将使用 default 命名空间。

对于 Pod 名称，如果没有明确提供，我们将使用 task_id。默认情况下会添加一个随机后缀，因此 Pod 名称通常无关紧要。

如何将集群 ConfigMaps、Secrets 和 Volumes 与 Pod 一起使用？¶

要添加 ConfigMaps、Volumes 和其他 Kubernetes 原生对象，我们建议您像这样导入 Kubernetes 模型 API

from kubernetes.client import models as k8s

使用此 API 对象，您可以以 Python 类的形式访问所有 Kubernetes API 对象。使用此方法将确保正确性和类型安全。虽然我们已经删除了几乎所有 Kubernetes 便利类，但我们保留了 Secret 类，以简化生成密钥卷/环境变量的过程。

tests/system/providers/cncf/kubernetes/example_kubernetes.py[源代码]

secret_file = Secret("volume", "/etc/sql_conn", "airflow-secrets", "sql_alchemy_conn")
secret_env = Secret("env", "SQL_CONN", "airflow-secrets", "sql_alchemy_conn")
secret_all_keys = Secret("env", None, "airflow-secrets-2")
volume_mount = k8s.V1VolumeMount(
    name="test-volume", mount_path="/root/mount_file", sub_path=None, read_only=True
)

configmaps = [
    k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name="test-configmap-1")),
    k8s.V1EnvFromSource(config_map_ref=k8s.V1ConfigMapEnvSource(name="test-configmap-2")),
]

volume = k8s.V1Volume(
    name="test-volume",
    persistent_volume_claim=k8s.V1PersistentVolumeClaimVolumeSource(claim_name="test-volume"),
)

port = k8s.V1ContainerPort(name="http", container_port=80)

init_container_volume_mounts = [
    k8s.V1VolumeMount(mount_path="/etc/foo", name="test-volume", sub_path=None, read_only=True)
]

init_environments = [k8s.V1EnvVar(name="key1", value="value1"), k8s.V1EnvVar(name="key2", value="value2")]

init_container = k8s.V1Container(
    name="init-container",
    image="ubuntu:16.04",
    env=init_environments,
    volume_mounts=init_container_volume_mounts,
    command=["bash", "-cx"],
    args=["echo 10"],
)

affinity = k8s.V1Affinity(
    node_affinity=k8s.V1NodeAffinity(
        preferred_during_scheduling_ignored_during_execution=[
            k8s.V1PreferredSchedulingTerm(
                weight=1,
                preference=k8s.V1NodeSelectorTerm(
                    match_expressions=[
                        k8s.V1NodeSelectorRequirement(key="disktype", operator="In", values=["ssd"])
                    ]
                ),
            )
        ]
    ),
    pod_affinity=k8s.V1PodAffinity(
        required_during_scheduling_ignored_during_execution=[
            k8s.V1WeightedPodAffinityTerm(
                weight=1,
                pod_affinity_term=k8s.V1PodAffinityTerm(
                    label_selector=k8s.V1LabelSelector(
                        match_expressions=[
                            k8s.V1LabelSelectorRequirement(key="security", operator="In", values="S1")
                        ]
                    ),
                    topology_key="failure-domain.beta.kubernetes.io/zone",
                ),
            )
        ]
    ),
)

tolerations = [k8s.V1Toleration(key="key", operator="Equal", value="value")]

Difference between `KubernetesPodOperator` and Kubernetes object spec ¶

可以将 KubernetesPodOperator 视为 Kubernetes 对象规范定义的替代品，该定义能够在 DAG 上下文中在 Airflow 调度程序中运行。如果使用该运算符，则无需为您要运行的 Pod 创建等效的 YAML/JSON 对象规范。仍然可以使用 pod_template_file 参数提供 YAML 文件，甚至可以使用需要 Kubernetes V1Pod 的 full_pod_spec 参数在 Python 中构造 Pod 规范。

如何使用私有镜像（容器注册表）？¶

默认情况下，KubernetesPodOperator 将在 Dockerhub 上查找公开托管的镜像。要从私有注册表（例如 ECR、GCR、Quay 或其他注册表）中提取镜像，您必须创建一个 Kubernetes Secret，该 Secret 表示用于从最终在 image_pull_secrets 参数中指定的私有注册表中访问镜像的凭据。

使用 kubectl 创建 Secret

kubectl create secret docker-registry testquay \
    --docker-server=quay.io \
    --docker-username=<Profile name> \
    --docker-password=<password>

然后像这样在您的 Pod 中使用它

tests/system/providers/cncf/kubernetes/example_kubernetes.py[源代码]

    quay_k8s = KubernetesPodOperator(
        namespace="default",
        image="quay.io/apache/bash",
        image_pull_secrets=[k8s.V1LocalObjectReference("testquay")],
        cmds=["bash", "-cx"],
        arguments=["echo", "10", "echo pwd"],
        labels={"foo": "bar"},
        name="airflow-private-image-pod",
        on_finish_action="delete_pod",
        in_cluster=True,
        task_id="task-two",
        get_logs=True,
    )

此外，对于此操作，您可以在可延迟模式下使用运算符

tests/system/providers/cncf/kubernetes/example_kubernetes_async.py[源代码]

    quay_k8s_async = KubernetesPodOperator(
        task_id="kubernetes_private_img_task_async",
        namespace="default",
        image="quay.io/apache/bash",
        image_pull_secrets=[k8s.V1LocalObjectReference("testquay")],
        cmds=["bash", "-cx"],
        arguments=["echo", "10", "echo pwd"],
        labels={"foo": "bar"},
        name="airflow-private-image-pod",
        on_finish_action="delete_pod",
        in_cluster=True,
        get_logs=True,
        deferrable=True,
    )

定期获取和显示容器日志的示例

tests/system/providers/cncf/kubernetes/example_kubernetes_async.py[源代码]

    kubernetes_task_async_log = KubernetesPodOperator(
        task_id="kubernetes_task_async_log",
        namespace="kubernetes_task_async_log",
        in_cluster=False,
        name="astro_k8s_test_pod",
        image="ubuntu",
        cmds=[
            "bash",
            "-cx",
            (
                "i=0; "
                "while [ $i -ne 100 ]; "
                "do i=$(($i+1)); "
                "echo $i; "
                "sleep 1; "
                "done; "
                "mkdir -p /airflow/xcom/; "
                'echo \'{"message": "good afternoon!"}\' > /airflow/xcom/return.json'
            ),
        ],
        do_xcom_push=True,
        deferrable=True,
        get_logs=True,
        logging_interval=5,
    )

XCom 如何工作？¶

KubernetesPodOperator 处理 XCom 值的方式与其他运算符不同。为了从您的 Pod 中传递 XCom 值，您必须将 do_xcom_push 指定为 True。这将创建一个与 Pod 一起运行的边车容器。Pod 必须将 XCom 值写入 /airflow/xcom/return.json 路径下的此位置。

注意

无效的 json 内容将失败，例如 echo 'hello' > /airflow/xcom/return.json 失败，而 echo '\"hello\"' > /airflow/xcom/return.json 成功

请参阅以下有关此过程的示例

tests/system/providers/cncf/kubernetes/example_kubernetes.py[源代码]

    write_xcom = KubernetesPodOperator(
        namespace="default",
        image="alpine",
        cmds=["sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json"],
        name="write-xcom",
        do_xcom_push=True,
        on_finish_action="delete_pod",
        in_cluster=True,
        task_id="write-xcom",
        get_logs=True,
    )

    pod_task_xcom_result = BashOperator(
        bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"",
        task_id="pod_task_xcom_result",
    )

    write_xcom >> pod_task_xcom_result

注意

仅针对标记为 State.SUCCESS 的任务推送 XCOM。

此外，对于此操作，您可以在可延迟模式下使用运算符

tests/system/providers/cncf/kubernetes/example_kubernetes_async.py[源代码]

    write_xcom_async = KubernetesPodOperator(
        task_id="kubernetes_write_xcom_task_async",
        namespace="default",
        image="alpine",
        cmds=["sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json"],
        name="write-xcom",
        do_xcom_push=True,
        on_finish_action="delete_pod",
        in_cluster=True,
        get_logs=True,
        deferrable=True,
    )

    pod_task_xcom_result_async = BashOperator(
        task_id="pod_task_xcom_result_async",
        bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"",
    )

    write_xcom_async >> pod_task_xcom_result_async

在电子邮件警报中包含错误消息 ¶

写入 /dev/termination-log 的任何内容都将由 Kubernetes 检索，并在任务失败时包含在异常消息中。

k = KubernetesPodOperator(
    task_id="test_error_message",
    image="alpine",
    cmds=["/bin/sh"],
    arguments=["-c", "echo hello world; echo Custom error > /dev/termination-log; exit 1;"],
    name="test-error-message",
    email="[email protected]",
    email_on_failure=True,
)

阅读有关 termination-log 的更多信息，请访问此处。

KubernetesPodOperator 回调 ¶

KubernetesPodOperator 支持不同的回调，这些回调可用于在 Pod 的生命周期内触发操作。为了使用它们，您需要创建一个 KubernetesPodOperatorCallback 的子类，并覆盖您要使用的回调方法。然后，您可以使用 callbacks 参数将您的回调类传递给操作符。

支持以下回调

on_sync_client_creation：在创建同步客户端后调用
on_pod_creation：在创建 Pod 后调用
on_pod_starting：在 Pod 启动后调用
on_pod_completion：在 Pod 完成时调用
on_pod_cleanup：在清理/删除 Pod 后调用
on_operator_resuming：从延迟状态恢复任务时调用
progress_callback：在容器日志的每一行调用

目前，在异步模式下不会调用回调方法，将来会添加此支持。

示例：¶

import kubernetes.client as k8s
import kubernetes_asyncio.client as async_k8s

from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from airflow.providers.cncf.kubernetes.callbacks import KubernetesPodOperatorCallback


class MyCallback(KubernetesPodOperatorCallback):
    @staticmethod
    def on_pod_creation(*, pod: k8s.V1Pod, client: k8s.CoreV1Api, mode: str, **kwargs) -> None:
        client.create_namespaced_service(
            namespace=pod.metadata.namespace,
            body=k8s.V1Service(
                metadata=k8s.V1ObjectMeta(
                    name=pod.metadata.name,
                    labels=pod.metadata.labels,
                    owner_references=[
                        k8s.V1OwnerReference(
                            api_version=pod.api_version,
                            kind=pod.kind,
                            name=pod.metadata.name,
                            uid=pod.metadata.uid,
                            controller=True,
                            block_owner_deletion=True,
                        )
                    ],
                ),
                spec=k8s.V1ServiceSpec(
                    selector=pod.metadata.labels,
                    ports=[
                        k8s.V1ServicePort(
                            name="http",
                            port=80,
                            target_port=80,
                        )
                    ],
                ),
            ),
        )


k = KubernetesPodOperator(
    task_id="test_callback",
    image="alpine",
    cmds=["/bin/sh"],
    arguments=["-c", "echo hello world; echo Custom error > /dev/termination-log; exit 1;"],
    name="test-callback",
    callbacks=MyCallback,
)

参考 ¶

如需了解更多信息，请查看

SparkKubernetesOperator ¶

SparkKubernetesOperator 允许您在 Kubernetes 集群上创建和运行 Spark 作业。它基于 spark-on-k8s-operator 项目。

此操作符简化了接口，并接受不同的参数来配置和运行 Kubernetes 上的 Spark 应用程序。与 KubernetesOperator 类似，我们添加了在提交后等待作业、管理错误处理、从驱动程序 Pod 检索日志以及删除 Spark 作业的功能的逻辑。它还支持开箱即用的 Kubernetes 功能，例如处理卷、配置映射、密钥等。

此操作符如何工作？¶

操作符通过在 Kubernetes 中生成 SparkApplication 自定义资源定义 (CRD) 来启动 Spark 任务。此 SparkApplication 任务随后使用用户指定的参数生成驱动程序和所需的执行程序 Pod。操作符会持续监控任务的进度，直到成功或失败。它从驱动程序 Pod 检索日志并将其显示在 Airflow UI 中。

使用示例 ¶

为了创建 SparkKubernetesOperator 任务，您必须提供一个包含 Spark 配置和 Kubernetes 相关资源配置的基本模板。此模板可以是 YAML 或 JSON 格式，作为操作符的起点。以下是您可以使用的示例模板

spark_job_template.yaml

spark:
  apiVersion: sparkoperator.k8s.io/v1beta2
  version: v1beta2
  kind: SparkApplication
  apiGroup: sparkoperator.k8s.io
  metadata:
    namespace: ds
  spec:
    type: Python
    pythonVersion: "3"
    mode: cluster
    sparkVersion: 3.0.0
    successfulRunHistoryLimit: 1
    restartPolicy:
      type: Never
    imagePullPolicy: Always
    hadoopConf: {}
    imagePullSecrets: []
    dynamicAllocation:
      enabled: false
      initialExecutors: 1
      minExecutors: 1
      maxExecutors: 1
    labels: {}
    driver:
      serviceAccount: default
      container_resources:
        gpu:
          name: null
          quantity: 0
        cpu:
          request: null
          limit: null
        memory:
          request: null
          limit: null
    executor:
      instances: 1
      container_resources:
        gpu:
          name: null
          quantity: 0
        cpu:
          request: null
          limit: null
        memory:
          request: null
          limit: null
kubernetes:
  # example:
  # env_vars:
  # - name: TEST_NAME
  #   value: TEST_VALUE
  env_vars: []

  # example:
  # env_from:
  # - name: test
  #   valueFrom:
  #     secretKeyRef:
  #       name: mongo-secret
  #       key: mongo-password
  env_from: []

  # example:
  # node_selector:
  #   karpenter.sh/provisioner-name: spark
  node_selector: {}

  # example: https://kubernetes.ac.cn/docs/concepts/scheduling-eviction/assign-pod-node/
  # affinity:
  #   nodeAffinity:
  #     requiredDuringSchedulingIgnoredDuringExecution:
  #       nodeSelectorTerms:
  #       - matchExpressions:
  #         - key: beta.kubernetes.io/instance-type
  #           operator: In
  #           values:
  #           - r5.xlarge
  affinity:
    nodeAffinity: {}
    podAffinity: {}
    podAntiAffinity: {}

  # example: https://kubernetes.ac.cn/docs/concepts/scheduling-eviction/taint-and-toleration/
  # type: list
  # tolerations:
  # - key: "key1"
  #   operator: "Equal"
  #   value: "value1"
  #   effect: "NoSchedule"
  tolerations: []

  # example:
  # config_map_mounts:
  #   snowflake-default: /mnt/tmp
  config_map_mounts: {}

  # example:
  # volume_mounts:
  # - name: config
  #   mountPath: /airflow
  volume_mounts: []

  # https://kubernetes.ac.cn/docs/concepts/storage/volumes/
  # example:
  # volumes:
  # - name: config
  #   persistentVolumeClaim:
  #     claimName: airflow
  volumes: []

  # read config map into an env variable
  # example:
  # from_env_config_map:
  # - configmap_1
  # - configmap_2
  from_env_config_map: []

  # load secret into an env variable
  # example:
  # from_env_secret:
  # - secret_1
  # - secret_2
  from_env_secret: []

  in_cluster: true
  conn_id: kubernetes_default
  kube_config_file: null
  cluster_context: null

重要

模板文件包含两个主要类别：spark 和 kubernetes。
- spark：此部分包含任务的 Spark 配置，镜像了 Spark API 模板的结构。
- kubernetes：此部分包含任务的 Kubernetes 资源配置，直接对应于 Kubernetes API 文档。每个资源类型在模板中都包含一个示例。
要使用的指定基础映像是 gcr.io/spark-operator/spark-py:v3.1.1。
确保 Spark 代码嵌入在镜像中、使用 persistentVolume 挂载或可从外部位置（如 S3 存储桶）访问。

接下来，使用以下内容创建任务

SparkKubernetesOperator(
    task_id="spark_task",
    image="gcr.io/spark-operator/spark-py:v3.1.1",  # OR custom image using that
    code_path="local://path/to/spark/code.py",
    application_file="spark_job_template.json",  # OR spark_job_template.json
    dag=dag,
)

注意：或者，application_file 也可以是 json 文件。请参见以下示例

spark_job_template.json

{
  "spark": {
    "apiVersion": "sparkoperator.k8s.io/v1beta2",
    "version": "v1beta2",
    "kind": "SparkApplication",
    "apiGroup": "sparkoperator.k8s.io",
    "metadata": {
      "namespace": "ds"
    },
    "spec": {
      "type": "Python",
      "pythonVersion": "3",
      "mode": "cluster",
      "sparkVersion": "3.0.0",
      "successfulRunHistoryLimit": 1,
      "restartPolicy": {
        "type": "Never"
      },
      "imagePullPolicy": "Always",
      "hadoopConf": {},
      "imagePullSecrets": [],
      "dynamicAllocation": {
        "enabled": false,
        "initialExecutors": 1,
        "minExecutors": 1,
        "maxExecutors": 1
      },
      "labels": {},
      "driver": {
        "serviceAccount": "default",
        "container_resources": {
          "gpu": {
            "name": null,
            "quantity": 0
          },
          "cpu": {
            "request": null,
            "limit": null
          },
          "memory": {
            "request": null,
            "limit": null
          }
        }
      },
      "executor": {
        "instances": 1,
        "container_resources": {
          "gpu": {
            "name": null,
            "quantity": 0
          },
          "cpu": {
            "request": null,
            "limit": null
          },
          "memory": {
            "request": null,
            "limit": null
          }
        }
      }
    }
  },
  "kubernetes": {
    "env_vars": [],
    "env_from": [],
    "node_selector": {},
    "affinity": {
      "nodeAffinity": {},
      "podAffinity": {},
      "podAntiAffinity": {}
    },
    "tolerations": [],
    "config_map_mounts": {},
    "volume_mounts": [
      {
        "name": "config",
        "mountPath": "/airflow"
      }
    ],
    "volumes": [
      {
        "name": "config",
        "persistentVolumeClaim": {
          "claimName": "hsaljoog-airflow"
        }
      }
    ],
    "from_env_config_map": [],
    "from_env_secret": [],
    "in_cluster": true,
    "conn_id": "kubernetes_default",
    "kube_config_file": null,
    "cluster_context": null
  }
}

除了使用 YAML 或 JSON 文件之外，另一种方法是直接传递 template_spec 字段而不是 application_file，如果您不想使用文件进行配置。

参考 ¶

如需了解更多信息，请查看

KubernetesJobOperator ¶

KubernetesJobOperator 允许您在 Kubernetes 集群上创建和运行作业。

注意

如果您使用托管 Kubernetes，请考虑使用专门的 KJO 操作符，因为它简化了 Kubernetes 授权过程

用于 Google Kubernetes Engine 的 GKEStartJobOperator 操作符。

注意

使用此运算符不需要 Kubernetes 执行器。

此操作符如何工作？¶

KubernetesJobOperator 使用 Kubernetes API 在 Kubernetes 集群中启动作业。操作符使用 Kube Python 客户端生成动态启动此作业的 Kubernetes API 请求。用户可以使用 config_file 参数指定 kubeconfig 文件，否则操作符将默认为 ~/.kube/config。它还允许用户使用 job_template_file 参数提供模板 YAML 文件。

tests/system/providers/cncf/kubernetes/example_kubernetes_job.py[源代码]

k8s_job = KubernetesJobOperator(
    task_id="job-task",
    namespace=JOB_NAMESPACE,
    image="perl:5.34.0",
    cmds=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"],
    name=JOB_NAME,
)

KubernetesJobOperator 还支持可延迟模式

tests/system/providers/cncf/kubernetes/example_kubernetes_job.py[源代码]

k8s_job_def = KubernetesJobOperator(
    task_id="job-task-def",
    namespace="default",
    image="perl:5.34.0",
    cmds=["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"],
    name=JOB_NAME + "-def",
    wait_until_job_complete=True,
    deferrable=True,
)

`KubernetesPodOperator` 和 `KubernetesJobOperator` 之间的区别 ¶

KubernetesJobOperator 是用于创建作业的操作符。作业会创建一个或多个 Pod，并将继续重试执行 Pod，直到指定数量的 Pod 成功终止。随着 Pod 成功完成，作业会跟踪成功的完成次数。当达到指定数量的成功完成次数时，作业即完成。用户可以使用配置参数（如 activeDeadlineSeconds 和 backoffLimit）来限制作业重试执行的次数。此操作符使用 KubernetesPodOperator 而不是 template 参数来创建 Pod。这意味着用户可以在 KubernetesJobOperator 中使用 KubernetesPodOperator 中的所有参数。

有关作业的更多信息，请访问：Kubernetes 作业文档

KubernetesDeleteJobOperator ¶

KubernetesDeleteJobOperator 允许您删除 Kubernetes 集群上的作业。

tests/system/providers/cncf/kubernetes/example_kubernetes_job.py[源代码]

delete_job_task = KubernetesDeleteJobOperator(
    task_id="delete_job_task",
    name=k8s_job.output["job_name"],
    namespace=JOB_NAMESPACE,
    wait_for_completion=True,
    delete_on_status="Complete",
    poll_interval=1.0,
)

KubernetesPatchJobOperator ¶

KubernetesPatchJobOperator 允许您更新 Kubernetes 集群上的作业。

tests/system/providers/cncf/kubernetes/example_kubernetes_job.py[源代码]

update_job = KubernetesPatchJobOperator(
    task_id="update-job-task",
    namespace="default",
    name=k8s_job.output["job_name"],
    body={"spec": {"suspend": False}},
)