Apache Airflow cncf.kubernetes Provider:Kubernetes 集群 Connection 的认证方式、参数详解与实现原理-二趣网

Apache Airflow cncf.kubernetes Provider:Kubernetes 集群 Connection 的认证方式、参数详解与实现原理

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

本文以 Airflowcncf.kubernetesprovider 的 Kubernetes 集群连接（Connection）文档为主线，完整讲解四种集群认证方式、kubernetes_default默认连接 ID、全部连接表单字段（in cluster、kube config 路径、JSON 格式 kube config、namespace、cluster context、SSL/TCP keepalive 开关及 XCom sidecar 三个扩展字段）、URI/JSON 两种环境变量注入示例，并结合 provider 源码（KubernetesHook）揭示配置的优先级判定逻辑与互斥约束，帮助你正确配置 Airflow 与 Kubernetes 集群之间的连接，并理解KubernetesPodOperator等任务消费该连接时的底层行为。

一、Kubernetes 集群 Connection 是什么

Kubernetes cluster Connection 类型让 Airflow 任务能够连接到 Kubernetes 集群。它主要服务于两类任务：

SparkKubernetesOperator（airflow.providers.cncf.kubernetes.operators.spark_kubernetes模块）；
KubernetesPodOperator（airflow.providers.cncf.kubernetes.operators.pod模块）。

在源码层面，该连接由 hooks/kubernetes.py 中的KubernetesHook承载，其类属性直接定义了连接的基本标识：

class KubernetesHook(BaseHook, PodOperatorHookProtocol): conn_name_attr = "kubernetes_conn_id" default_conn_name = "kubernetes_default" conn_type = "kubernetes" hook_name = "Kubernetes Cluster Connection"

可以看到三个关键事实：

连接的conn_type为kubernetes，即 URI 格式中kubernetes://...的 scheme 来源；
默认连接 ID 为kubernetes_default，与文档"Default Connection IDs"一节一致；
任务中通常通过kubernetes_conn_id参数传入具体连接。

二、四种集群认证方式

文档给出了 Airflow 连接 Kubernetes 的四种途径，按推荐场景排序：

使用默认位置的 kube_config：kube config 位于机器的默认位置~/.kube/config时，所有连接字段留空即可；
使用 in cluster 配置：当 Airflow 运行在 Kubernetes 集群内部时，标记 "In cluster configuration" 选项，直接采用集群内凭证（ServiceAccount）；
使用其他位置的 kube_config：把自定义路径填入Kube config path字段；
使用 JSON 格式的 kube_config：把 kubeconfig 内容直接粘贴进Kube config (JSON format)字段。

源码视角:配置的判定顺序与互斥校验

KubernetesHook的类文档字符串明确说明了判定顺序（hooks/kubernetes.py）："use in cluster configuration... use custom config by providing path... use custom configuration by providing content of kubeconfig file via extra fieldkube_config... use default config by providing no extras"。

get_conn()方法（hooks/kubernetes.py）实现了这一逻辑，并对四种来源做了互斥校验：

num_selected_configuration = sum( 1 for o in [in_cluster, kubeconfig, kubeconfig_path, self.config_dict] if o ) if num_selected_configuration > 1: raise AirflowException( "Invalid connection configuration. Options kube_config_path, " "kube_config, in_cluster, config_dict are mutually exclusive. " "You can only use one option at a time." )

也就是说，同一个连接中in_cluster、kube_config_path、kube_config、config_dict四个选项最多只能生效一个，否则会抛出AirflowException。各分支的实际行为是：

in_cluster为真时调用config.load_incluster_config()，使用集群内 ServiceAccount 凭证；
kube_config_path存在时调用config.load_kube_config(config_file=..., context=cluster_context)，支持按 context 选择集群上下文；
kube_config（JSON/字符串）存在时，先写入临时文件再调用config.load_kube_config()加载；
以上都未提供时，走_get_default_client()：先尝试 in-cluster 配置，失败（ConfigException）后回退到默认位置的~/.kube/config——这正是文档第 1 种"所有字段留空"方式的底层实现。

此外还有一个实用细节：get_connection()类方法（hooks/kubernetes.py）在连接不存在且conn_id == "kubernetes_default"时会返回一个空连接，让 hook 回落到集群派生的凭证，因此在集群内运行时即使没有显式创建kubernetes_default连接也能正常工作。

三、默认连接 ID

默认连接 ID 是kubernetes_default。这一点与源码中的default_conn_name = "kubernetes_default"一致，并且只有这个 ID 享有"连接缺失时返回空连接"的兜底行为——其他自定义conn_id若不存在会正常抛出AirflowNotFoundException。

四、连接表单字段详解

UI 中隐藏了host、schema、login、password、port、extra这些通用字段（见get_ui_field_behaviour()，hooks/kubernetes.py），只暴露以下专用字段。各字段在get_connection_form_widgets()中定义，字段名即extra字典中的键名。

in cluster configuration

使用 in cluster 配置。对应 extra 键in_cluster。当 Airflow 组件（如 worker、triggerer）本身运行在集群中且拥有合适 ServiceAccount 权限时，勾选此项即可免去任何 kubeconfig 文件。

Kube config path

使用自定义路径的 kube config。对应 extra 键kube_config_path，支持~展开（如~/.kube/config）。

Kube config (JSON format)

用于连接 Kubernetes 客户端的 Kube config，可以直接把 kubeconfig 内容（JSON 格式）粘贴到该字段。对应 extra 键kube_config，UI 上以密码框（PasswordField）呈现，避免内容在表单中明文回显。

Namespace

该连接使用的默认 Kubernetes namespace。对应 extra 键namespace；KubernetesHook中定义了常量DEFAULT_NAMESPACE = "default"，即未指定时任务默认落在default命名空间。

Cluster context

使用 kube config 时可指定使用哪个 context（多集群场景下 kubeconfig 中可能包含多个 context）。对应 extra 键cluster_context，最终传给load_kube_config(context=...)。

Disable verify SSL

可选地禁用 SSL 证书校验。对应 extra 键disable_verify_ssl，默认 SSL 是被校验的。源码中get_conn()里若该值为True会调用_disable_verify_ssl()；对应地，只有当disable_tcp_keepalive不为True时才会调用_enable_tcp_keepalive()。

Disable TCP keepalive

TCP keepalive 是一个默认启用的特性，用于保持长连接存活。将其设为True可禁用该特性。对应 extra 键disable_tcp_keepalive，适用于某些代理/负载均衡环境下 keepalive 探测导致连接被异常重置的场景。

Xcom sidecar image

定义PodDefaults.SIDECAR_CONTAINER使用的image，默认"alpine"，可用于指向私有镜像仓库或自定义镜像覆盖。对应 extra 键xcom_sidecar_container_image。

从 utils/xcom_sidecar.py 的源码可以看到，XCom sidecar 的镜像实际被固定版本号：

XCOM_SIDECAR_IMAGE = "alpine:3.24.1"

源码注释解释了固定版本而非:latest的原因：固定 tag 会让 kubelet 默认的imagePullPolicy=IfNotPresent生效，已缓存镜像的节点不会在每次任务时重新拉取，从而避免匿名拉取 Docker Hub 的限流问题，同时保护 CI 与离线部署环境。因此，如果你的集群无法匿名访问 Docker Hub，可通过该字段覆盖为私有仓库中的对应镜像。

Xcom sidecar resources (JSON format)

以 JSON 对象形式为 XCom sidecar 容器定义资源requests/limits，例如：

{"requests": {"cpu": "1m", "memory": "10Mi"}}

对应 extra 键xcom_sidecar_container_resources。用于满足集群的资源配额（ResourceQuota）或对 sidecar 做最小化资源约束。

Xcom sidecar security context (JSON format)

以 JSON 对象形式为 XCom sidecar 容器定义securityContext，例如：

{"allowPrivilegeEscalation": false, "readOnlyRootFilesystem": true, "seccompProfile": {"type": "RuntimeDefault"}}

对应 extra 键xcom_sidecar_container_security_context。当集群对注入的 sidecar 强制执行 Pod Security Standards 或准入策略（如 OPA/Gatekeeper）时非常有用。需要区分的是：KubernetesPodOperator的 DAG 作者仍可通过任务级参数xcom_sidecar_container_security_context对单个任务进行覆盖，连接级配置提供的是全局默认值。

五、以环境变量方式存储连接:URI 与 JSON 两种格式

URI 格式

文档给出的环境变量示例：

AIRFLOW_CONN_KUBERNETES_DEFAULT='kubernetes://?in_cluster=True&kube_config_path=~%2F.kube%2Fconfig&kube_config=kubeconfig+json&namespace=namespace'

注意其中路径的分隔符经过了 URL 编码（~%2F.kube%2Fconfig即~/.kube/config）。该示例同时展示了多个字段的拼接方式；结合第二节的互斥校验，实际使用时请只保留一种集群来源选项。

JSON 格式

AIRFLOW_CONN_KUBERNETES_DEFAULT='{"conn_type": "kubernetes", "extra": {"in_cluster": true, "kube_config_path": "~/.kube/config", "namespace": "my-namespace"}}'

JSON 格式下conn_type必须是kubernetes，所有专用字段都放在extra中，键名与 UI 字段名一致（in_cluster、kube_config_path、kube_config、namespace、cluster_context、disable_verify_ssl、disable_tcp_keepalive、xcom_sidecar_container_image、xcom_sidecar_container_resources、xcom_sidecar_container_security_context）。

兼容性提示:extra 字段的前缀回退

如果你维护的是较老版本的连接数据，_get_field()方法（hooks/kubernetes.py）提供了向后兼容：Airflow 2.3 之前，extra 字段曾需要extra__kubernetes__前缀存储，该方法会先查找无前缀字段、再回退到带前缀字段。因此旧格式的 extra 数据无需迁移即可继续工作，但新写入的连接应使用不带前缀的字段名。

六、小结:配置 Kubernetes 连接的实践路径

结合本文档与源码，可以归纳出一条清晰的配置决策路径：

集群内运行 Airflow 且组件有 ServiceAccount 权限：勾选In cluster configuration（in_cluster=True）即可，其余留空；也可以完全不创建连接，依赖kubernetes_default的兜底逻辑自动尝试 in-cluster 配置；
本机开发、kubeconfig 在默认位置：所有字段留空，hook 会自动尝试 in-cluster 失败后回落到~/.kube/config；
kubeconfig 在自定义路径：填Kube config path；
无法挂载文件、希望把配置直接存进 Airflow 元数据库：使用Kube config (JSON format)字段，或AIRFLOW_CONN_KUBERNETES_DEFAULT环境变量注入；
多集群 kubeconfig：追加Cluster context指定上下文；
集群有安全准入策略或私有镜像仓库：按需配置 XCom sidecar 的image、resources、securityContext三个字段，使注入的 sidecar 容器满足 Pod Security Standards 并通过资源配额校验。

参考文档：Kubernetes cluster Connection 原文档；核心实现见 KubernetesHook 与 XCom sidecar 工具。

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析