airflow.providers.cncf.kubernetes.operators.pod¶

在 Kubernetes POD 中執行任務。

屬性¶

`alphanum_lower`
`KUBE_CONFIG_ENV_VAR`

異常¶

`PodReattachFailure`	當我們期望能夠找到一個 Pod 但找不到時。
`PodCredentialsExpiredFailure`	當 Pod 重新整理憑據失敗時。

類¶

`PodEventType`	Kubernetes Pod 發出的事件型別。
`KubernetesPodOperator`	在 Kubernetes Pod 中執行任務。

模組內容¶

airflow.providers.cncf.kubernetes.operators.pod.alphanum_lower = 'abcdefghijklmnopqrstuvwxyz0123456789'[原始碼]¶

airflow.providers.cncf.kubernetes.operators.pod.KUBE_CONFIG_ENV_VAR = 'KUBECONFIG'[原始碼]¶

class airflow.providers.cncf.kubernetes.operators.pod.PodEventType[原始碼]¶

繼承自: enum.Enum

Kubernetes Pod 發出的事件型別。

WARNING = 'Warning'[原始碼]¶

NORMAL = 'Normal'[原始碼]¶

exception airflow.providers.cncf.kubernetes.operators.pod.PodReattachFailure[原始碼]¶

繼承自: airflow.exceptions.AirflowException

當我們期望能夠找到一個 Pod 但找不到時。

exception airflow.providers.cncf.kubernetes.operators.pod.PodCredentialsExpiredFailure[原始碼]¶

繼承自: airflow.exceptions.AirflowException

當 Pod 重新整理憑據失敗時。

class airflow.providers.cncf.kubernetes.operators.pod.KubernetesPodOperator(*, kubernetes_conn_id=KubernetesHook.default_conn_name, namespace=None, image=None, name=None, random_name_suffix=True, cmds=None, arguments=None, ports=None, volume_mounts=None, volumes=None, env_vars=None, env_from=None, secrets=None, in_cluster=None, cluster_context=None, labels=None, reattach_on_restart=True, startup_timeout_seconds=120, startup_check_interval_seconds=5, get_logs=True, base_container_name=None, base_container_status_polling_interval=1, init_container_logs=None, container_logs=None, image_pull_policy=None, annotations=None, container_resources=None, affinity=None, config_file=None, node_selector=None, image_pull_secrets=None, service_account_name=None, hostnetwork=False, host_aliases=None, tolerations=None, security_context=None, container_security_context=None, dnspolicy=None, dns_config=None, hostname=None, subdomain=None, schedulername=None, full_pod_spec=None, init_containers=None, log_events_on_failure=False, do_xcom_push=False, pod_template_file=None, pod_template_dict=None, priority_class_name=None, pod_runtime_info_envs=None, termination_grace_period=None, configmaps=None, skip_on_exit_code=None, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), poll_interval=2, log_pod_spec_on_failure=True, on_finish_action='delete_pod', is_delete_operator_pod=None, termination_message_policy='File', active_deadline_seconds=None, callbacks=None, progress_callback=None, logging_interval=None, **kwargs)[原始碼]¶

繼承自: airflow.models.BaseOperator

在 Kubernetes Pod 中執行任務。

另請參閱

有關如何使用此 Operator 的更多資訊，請參閱指南: KubernetesPodOperator

注意

如果您使用 Google Kubernetes Engine 且 Airflow 未在同一叢集中執行，請考慮使用 GKEStartPodOperator，它簡化了授權過程。

引數:

kubernetes_conn_id (str | None) – 用於 Kubernetes 叢集的 kubernetes 連線 ID。(模板化)
namespace (str | None) – 在 kubernetes 內執行的名稱空間。
image (str | None) – 您希望啟動的容器映象。預設為 hub.docker.com，但完全限定的 URL 將指向自定義倉庫。(模板化)
name (str | None) – 任務執行所在的 Pod 的名稱，將用於（如果 random_name_suffix 為 True，則加上隨機字尾）生成 Pod ID（DNS-1123 子域名，僅包含 [a-z0-9.-]）。(模板化)
random_name_suffix (bool) – 如果為 True，將生成隨機字尾。
cmds (list[str] | None) – 容器的入口點。如果未提供，則使用容器映象的入口點。(模板化)
arguments (list[str] | None) – 入口點的引數。如果未提供，則使用容器映象的 CMD。(模板化)
ports (list[kubernetes.client.models.V1ContainerPort] | None) – 啟動的 Pod 的埠。
volume_mounts (list[kubernetes.client.models.V1VolumeMount] | None) – 啟動的 Pod 的卷掛載。(模板化)
volumes (list[kubernetes.client.models.V1Volume] | None) – 啟動的 Pod 的卷。包括 ConfigMaps 和 PersistentVolumes。(模板化)
env_vars (list[kubernetes.client.models.V1EnvVar] | dict[str, str] | None) – 在容器中初始化的環境變數。(模板化)
env_from (list[kubernetes.client.models.V1EnvFromSource] | None) – (可選) 用於填充容器中環境變數的來源列表。(模板化)
secrets (list[airflow.providers.cncf.kubernetes.secret.Secret] | None) – 要注入到容器中的 Kubernetes Secret。它們可以作為環境變數或卷中的檔案暴露。
in_cluster (bool | None) – 使用 in_cluster 配置執行 Kubernetes 客戶端。
cluster_context (str | None) – 指向 Kubernetes 叢集的上下文。當 in_cluster 為 True 時忽略。如果為 None，則使用當前上下文。(模板化)
reattach_on_restart (bool) – 如果 Pod 執行時 worker 死亡，則在下次嘗試時重新連線並監控。如果為 False，則每次嘗試都建立一個新的 Pod。
labels (dict | None) – 應用於 Pod 的標籤。(模板化)
startup_timeout_seconds (int) – Pod 啟動的超時時間（秒）。
startup_check_interval_seconds (int) – 檢查 Pod 是否已啟動的間隔（秒）
get_logs (bool) – 將基礎容器的 stdout 作為任務日誌獲取。
init_container_logs (collections.abc.Iterable[str] | str | Literal[True] | None) – init 容器列表，其日誌將釋出到 stdout。接受容器序列、單個容器名稱或 True。如果為 True，則所有容器的日誌都會發布。
container_logs (collections.abc.Iterable[str] | str | Literal[True] | None) – 容器列表，其日誌將釋出到 stdout。接受容器序列、單個容器名稱或 True。如果為 True，則所有容器的日誌都會發布。與 get_logs 引數配合使用。預設值為基礎容器。
image_pull_policy (str | None) – 指定快取或始終拉取映象的策略。
annotations (dict | None) – 可以附加到 Pod 的非識別元資料。可以包含大量資料，並且可以包含標籤不允許的字元。(模板化)
container_resources (kubernetes.client.models.V1ResourceRequirements | None) – 啟動的 Pod 的資源。(模板化)
affinity (kubernetes.client.models.V1Affinity | None) – 啟動的 Pod 的親和性排程規則。
config_file (str | None) – Kubernetes 配置檔案的路徑。(模板化) 如果未指定，預設值為 ~/.kube/config
node_selector (dict | None) – 包含一組排程規則的字典。(模板化)
image_pull_secrets (list[kubernetes.client.models.V1LocalObjectReference] | None) – 需要提供給 Pod 的所有映象拉取 Secret。如果需要多個 Secret，請提供一個逗號分隔列表：secret_a,secret_b
service_account_name (str | None) – 服務賬號的名稱
hostnetwork (bool) – 如果為 True，則在 Pod 上啟用主機網路。
host_aliases (list[kubernetes.client.models.V1HostAlias] | None) – 應用於 Pod 中容器的主機別名列表。
tolerations (list[kubernetes.client.models.V1Toleration] | None) – Kubernetes tolerations（容忍度）列表。
security_context (kubernetes.client.models.V1PodSecurityContext | dict | None) – Pod 應執行的安全選項 (PodSecurityContext)。
container_security_context (kubernetes.client.models.V1SecurityContext | dict | None) – 容器應執行的安全選項。
dnspolicy (str | None) – Pod 的 dnspolicy（DNS 策略）。
dns_config (kubernetes.client.models.V1PodDNSConfig | None) – Pod 的 DNS 配置（IP 地址、搜尋域、選項）。
hostname (str | None) – Pod 的主機名。(模板化)
subdomain (str | None) – Pod 的子域名。
schedulername (str | None) – 指定 Pod 的排程器名稱。
full_pod_spec (kubernetes.client.models.V1Pod | None) – 完整的 PodSpec。
init_containers (list[kubernetes.client.models.V1Container] | None) – 啟動的 Pod 的 init 容器。
log_events_on_failure (bool) – 如果發生故障，記錄 Pod 的事件日誌。
do_xcom_push (bool) – 如果為 True，容器中檔案 /airflow/xcom/return.json 的內容也將在容器完成後被推送到 XCom。
pod_template_file (str | None) – Pod 模板檔案的路徑。(模板化)
pod_template_dict (dict | None) – Pod 模板字典。(模板化)
priority_class_name (str | None) – 啟動的 Pod 的優先順序類名稱。
pod_runtime_info_envs (list[kubernetes.client.models.V1EnvVar] | None) – (可選) 一組環境變數，用於設定在容器中。
termination_grace_period (int | None) – 如果任務在 UI 中被終止，優雅停機期，預設為 kubernetes 預設值。
configmaps (list[str] | None) – (可選) ConfigMap 名稱列表，從中收集 ConfigMap 以環境變數填充。目標 ConfigMap 的 Data 欄位內容將表示為環境變數的鍵值對。擴充套件 env_from。
skip_on_exit_code (int | collections.abc.Container[int] | None) – 如果任務以此退出程式碼退出，則將任務保持在 skipped 狀態（預設：None）。如果設定為 None，任何非零退出程式碼都將被視為失敗。
base_container_name (str | None) – Pod 中的基礎容器名稱。如果 get_logs 為 True，此容器的日誌將作為任務日誌的一部分顯示。預設為 None。如果為 None，將查閱類變數 BASE_CONTAINER_NAME（預設為“base”）作為要使用的基礎容器名稱。(模板化)
base_container_status_polling_interval (float) – 輪詢週期（秒），用於檢查 Pod 基礎容器狀態。預設為 1 秒。
deferrable (bool) – 在可延遲模式下執行 Operator。
poll_interval (float) – 檢查狀態的輪詢週期（秒）。僅在可延遲模式下使用。
log_pod_spec_on_failure (bool) – 如果發生故障，記錄 Pod 的規範。
on_finish_action (str) – 當 Pod 達到其最終狀態或執行中斷時要做什麼。如果為“delete_pod”，無論狀態如何，Pod 都將被刪除；如果為“delete_succeeded_pod”，則僅刪除成功的 Pod。可以設定為“keep_pod”來保留 Pod。
termination_message_policy (str) – 基礎容器的終止訊息策略。預設值為“File”。
active_deadline_seconds (int | None) – active_deadline_seconds，它轉換為 V1PodSpec 中的 active_deadline_seconds。
callbacks (list[type[airflow.providers.cncf.kubernetes.callbacks.KubernetesPodOperatorCallback]] | type[airflow.providers.cncf.kubernetes.callbacks.KubernetesPodOperatorCallback] | None) – 包含 KubernetesPodOperator 不同步驟的回撥方法的 KubernetesPodOperatorCallback 例項。
logging_interval (int | None) – 任務在恢復獲取最新日誌之前可以處於延遲狀態的最大時間（秒）。如果為 None，則任務將保持延遲狀態直到 Pod 完成，在此之前不會顯示任何日誌。

BASE_CONTAINER_NAME = 'base'[原始碼]¶

ISTIO_CONTAINER_NAME = 'istio-proxy'[原始碼]¶

KILL_ISTIO_PROXY_SUCCESS_MSG = 'HTTP/1.1 200'[原始碼]¶

POD_CHECKED_KEY = 'already_checked'[原始碼]¶

POST_TERMINATION_TIMEOUT = 120[source]¶

template_fields: collections.abc.Sequence[str] = ('image', 'name', 'hostname', 'cmds', 'annotations', 'arguments', 'env_vars', 'labels',...[source]¶

template_fields_renderers[source]¶

kubernetes_conn_id = 'kubernetes_default'[source]¶

do_xcom_push = False[source]¶

image = None[source]¶

namespace = None[source]¶

cmds = [][source]¶

arguments = [][source]¶

labels[source]¶

startup_timeout_seconds = 120[source]¶

startup_check_interval_seconds = 5[source]¶

env_vars = [][source]¶

pod_runtime_info_envs = [][source]¶

env_from = [][source]¶

ports = [][source]¶

volume_mounts = [][source]¶

volumes = [][source]¶

secrets = [][source]¶

in_cluster = None[source]¶

cluster_context = None[source]¶

reattach_on_restart = True[source]¶

get_logs = True[source]¶

base_container_name = 'base'[source]¶

base_container_status_polling_interval = 1[source]¶

init_container_logs = None[source]¶

container_logs = 'base'[source]¶

image_pull_policy = None[source]¶

node_selector[source]¶

annotations[source]¶

affinity[source]¶

container_resources = None[source]¶

config_file = None[source]¶

image_pull_secrets = [][source]¶

service_account_name = None[source]¶

hostnetwork = False[source]¶

host_aliases = None[source]¶

tolerations = [][source]¶

security_context[source]¶

container_security_context = None[source]¶

dnspolicy = None[source]¶

dns_config = None[source]¶

hostname = None[source]¶

subdomain = None[source]¶

schedulername = None[source]¶

full_pod_spec = None[source]¶

init_containers = [][source]¶

log_events_on_failure = False[source]¶

priority_class_name = None[source]¶

pod_template_file = None[source]¶

pod_template_dict = None[source]¶

name = None[source]¶

random_name_suffix = True[source]¶

termination_grace_period = None[source]¶

pod_request_obj: kubernetes.client.models.V1Pod | None = None[source]¶

pod: kubernetes.client.models.V1Pod | None = None[source]¶

skip_on_exit_code = None[source]¶

deferrable = True[source]¶

poll_interval = 2[source]¶

remote_pod: kubernetes.client.models.V1Pod | None = None[source]¶

log_pod_spec_on_failure = True[source]¶

on_finish_action[source]¶

termination_message_policy = 'File'[source]¶

active_deadline_seconds = None[source]¶

logging_interval = None[source]¶

callbacks = [][source]¶

property pod_manager: airflow.providers.cncf.kubernetes.utils.pod_manager.PodManager[source]¶

property hook: airflow.providers.cncf.kubernetes.utils.pod_manager.PodOperatorHookProtocol[source]¶

property client: kubernetes.client.CoreV1Api[source]¶

find_pod(namespace, context, *, exclude_checked=True)[source]¶

如果存在，返回此任務例項已在執行的 Pod。

log_matching_pod(pod, context)[source]¶

get_or_create_pod(pod_request_obj, context)[source]¶

await_pod_start(pod)[source]¶

extract_xcom(pod)[source]¶

檢索 xcom 值並終止 xcom 邊車容器。

execute(context)[source]¶

根據 deferrable 引數，非同步或同步執行 Pod。

execute_sync(context)[source]¶

await_init_containers_completion(pod)[source]¶

await_pod_completion(pod)[source]¶

execute_async(context)[source]¶

convert_config_file_to_dict()[source]¶: 將傳入的 config_file 轉換為字典表示形式。

invoke_defer_method(last_log_time=None)[source]¶

重新定義子類中使用的觸發器。

trigger_reentry(context, event)[source]¶

從觸發器返回的入口點。

如果 logging_interval 為 None，則此時 Pod 應該已完成，我們將只需獲取日誌並退出。

如果 logging_interval 不為 None，則 Pod 可能仍在執行，我們將只需獲取最新日誌並再次延遲迴觸發器。

post_complete_action(*, pod, remote_pod, context, **kwargs)[source]¶

運算子完成可延遲執行邏輯後必須執行的操作。

cleanup(pod, remote_pod)[source]¶

is_istio_enabled(pod)[source]¶

透過檢查名稱空間標籤來檢查 Pod 的名稱空間是否啟用了 Istio。

kill_istio_sidecar(pod)[source]¶

process_pod_deletion(pod, *, reraise=True)[source]¶

patch_already_checked(pod, *, reraise=True)[source]¶

新增一個“已檢查”標籤，以確保在重試時不再重新連線。

on_kill()[source]¶

覆蓋此方法以在任務例項被終止時清理子程序。

在運算子內使用 threading、subprocess 或 multiprocessing 模組都需要進行清理，否則會留下殘留程序。

build_pod_request_obj(context=None)[source]¶

根據 Pod 模板檔案、完整的 Pod 規範以及其他運算子引數返回 V1Pod 物件。

V1Pod 屬性（按優先順序）從運算子引數、完整的 Pod 規範、Pod 模板檔案派生而來。

dry_run()[source]¶

列印此運算子將建立的 Pod 定義。

不包括任務例項特有的標籤（因為模擬執行中不存在）並排除所有空元素。

process_duplicate_label_pods(pod_list)[source]¶

修補或刪除帶有重複標籤的現有 Pod。

這是為了處理一種邊緣情況，只有當 reattach_on_restart 標誌為 False 且上一次執行嘗試因任務程序被叢集或另一個程序外部終止而失敗時才會發生。

如果任務程序被外部終止，會中斷程式碼執行並立即退出任務。結果是，在上一次嘗試中建立的 Pod 將不會被 cleanup() 方法正確刪除或修補。

返回用於下一次執行嘗試的新建立的 Pod。