使用 KubernetesPodOperator

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

本頁說明如何使用 KubernetesPodOperator，將 Cloud Composer 中的 Kubernetes Pod 部署到 Cloud Composer 環境的 Google Kubernetes Engine 叢集。

KubernetesPodOperator 會在環境的叢集中啟動 Kubernetes Pod。相較之下，Google Kubernetes Engine 運算子會在指定叢集中執行 Kubernetes Pod，這個叢集可能與您的環境無關。您也可以使用 Google Kubernetes Engine 運算子建立及刪除叢集。

如果您需要下列功能，KubernetesPodOperator 會是不錯的選擇：

• 無法透過公開 PyPI 存放區取得的自訂 Python 依附元件。
• Cloud Composer 工作站映像檔中沒有的二進位依附元件。

注意： 如要進一步瞭解如何安裝私人套件存放區中託管的套件，或管理依附元件的其他選項，請參閱「安裝 Python 依附元件」。

事前準備

請查看下列 Cloud Composer 3 和 Cloud Composer 2 中 KubernetesPodOperator 的差異清單，並確認 DAG 相容：

• 您無法在 Cloud Composer 3 中建立自訂命名空間。即使指定了其他命名空間，Pod 也一律會在 composer-user-workloads 命名空間中執行。這個命名空間中的 Pod 可存取專案資源和 VPC 網路 (如已啟用)，不需額外設定。

• Cloud Composer 3 無法執行多個額外的 Sidecar 容器。如果側車容器命名為 airflow-xcom-sidecar，您就可以執行該容器。

• 無法使用 Kubernetes API 建立 Kubernetes Secret 和 ConfigMap。 Cloud Composer 提供 Google Cloud CLI 指令、Terraform 資源和 Cloud Composer API，可管理 Kubernetes Secrets 和 ConfigMap。詳情請參閱「使用 Kubernetes Secret 和 ConfigMap」。

• 您無法在 Cloud Composer 3 中部署自訂工作負載。您只能修改 Kubernetes Secret 和 ConfigMap，無法進行其他設定變更。

• 資源需求 (CPU、記憶體和儲存空間) 必須使用支援的值指定。

• 與 Cloud Composer 2 相同，Pod 親和性設定不適用於 Cloud Composer 1。如要使用 Pod 相依性，請改用 GKE 運算子在其他叢集中啟動 Pod。

Cloud Composer 3 中的 KubernetesPodOperator 簡介

本節說明 KubernetesPodOperator 在 Cloud Composer 3 中的運作方式。

資源使用情況

在 Cloud Composer 3 中，環境的叢集會自動調整資源配置。您使用 KubernetesPodOperator 執行的額外工作負載，會與環境分開擴充。資源需求增加不會影響您的環境，但環境的叢集會視資源需求擴大或縮小。

在環境叢集中執行的額外工作負載，其定價會採用 Cloud Composer 3 定價模式，並使用 Cloud Composer 3 SKU。

Cloud Composer 3 使用 Autopilot 叢集，並導入運算級別的概念：

• Cloud Composer 僅支援 general-purpose 運算類別。

• 根據預設，如果未選取任何類別，使用 KubernetesPodOperator 建立 Pod 時，系統會假設為 general-purpose 類別。

• 每個類別都與特定屬性和資源限制相關聯，詳情請參閱 Autopilot 說明文件。舉例來說，在 general-purpose 類別中執行的 Pod 最多可使用 110 GiB 的記憶體。

專案資源的存取權

在 Cloud Composer 3 中，環境的叢集位於租戶專案中，無法進行設定。Pod 會在環境的叢集中執行，並位於獨立的命名空間。

在 Cloud Composer 3 中，即使指定了其他命名空間，Pod 也一律會在 composer-user-workloads命名空間中執行。這個命名空間中的 Pod 可以存取專案中的資源和虛擬私有雲網路 (如果已啟用)，不需額外設定。 Google Cloud環境的服務帳戶會用於存取這些資源。您無法指定其他服務帳戶。

希望將設定需求減至最低

如要建立 KubernetesPodOperator，只需要 Pod 的 name、要使用的 image 和 task_id 參數。/home/airflow/composer_kube_config 包含向 GKE 進行驗證的憑證。

kubernetes_min_pod = KubernetesPodOperator(
    # The ID specified for the task.
    task_id="pod-ex-minimum",
    # Name of task you want to run, used to generate Pod ID.
    name="pod-ex-minimum",
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["echo"],
    # The namespace to run within Kubernetes. In Composer 2 environments
    # after December 2022, the default namespace is
    # `composer-user-workloads`. Always use the
    # `composer-user-workloads` namespace with Composer 3.
    namespace="composer-user-workloads",
    # Docker image specified. Defaults to hub.docker.com, but any fully
    # qualified URLs will point to a custom repository. Supports private
    # gcr.io images if the Composer Environment is under the same
    # project-id as the gcr.io images and the service account that Composer
    # uses has permission to access the Google Container Registry
    # (the default service account has permission)
    image="gcr.io/gcp-runtimes/ubuntu_20_0_4",
    # Specifies path to kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
)

額外設定

這個範例顯示您可以在 KubernetesPodOperator 中設定的其他參數。

詳情請參閱下列資源：

• 如要瞭解如何使用 Kubernetes Secret 和 ConfigMap，請參閱「使用 Kubernetes Secret 和 ConfigMap」。

• 如要瞭解如何搭配 KubernetesPodOperator 使用 Jinja 範本，請參閱「使用 Jinja 範本」。

• 如要瞭解資源需求 (CPU、記憶體和儲存空間) 的支援值，請參閱資源需求。

• 如要瞭解 KubernetesPodOperator 參數，請參閱 Airflow 說明文件中的運算子參考資料。

kubernetes_full_pod = KubernetesPodOperator(
    task_id="ex-all-configs",
    name="pi",
    namespace="composer-user-workloads",
    image="perl:5.34.0",
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["perl"],
    # Arguments to the entrypoint. The Docker image's CMD is used if this
    # is not provided. The arguments parameter is templated.
    arguments=["-Mbignum=bpi", "-wle", "print bpi(2000)"],
    # The secrets to pass to Pod, the Pod will fail to create if the
    # secrets you specify in a Secret object do not exist in Kubernetes.
    secrets=[],
    # Labels to apply to the Pod.
    labels={"pod-label": "label-name"},
    # Timeout to start up the Pod, default is 600.
    startup_timeout_seconds=600,
    # The environment variables to be initialized in the container.
    # The env_vars parameter is templated.
    env_vars={"EXAMPLE_VAR": "/example/value"},
    # If true, logs stdout output of container. Defaults to True.
    get_logs=True,
    # Determines when to pull a fresh image, if 'IfNotPresent' will cause
    # the Kubelet to skip pulling an image if it already exists. If you
    # want to always pull a new image, set it to 'Always'.
    image_pull_policy="Always",
    # Annotations are non-identifying metadata you can attach to the Pod.
    # Can be a large range of data, and can include characters that are not
    # permitted by labels.
    annotations={"key1": "value1"},
    # Optional resource specifications for Pod, this will allow you to
    # set both cpu and memory limits and requirements.
    # Prior to Airflow 2.3 and the cncf providers package 5.0.0
    # resources were passed as a dictionary. This change was made in
    # https://github.com/apache/airflow/pull/27197
    # Additionally, "memory" and "cpu" were previously named
    # "limit_memory" and "limit_cpu"
    # resources={'limit_memory': "250M", 'limit_cpu': "100m"},
    container_resources=k8s_models.V1ResourceRequirements(
        requests={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
        limits={"cpu": "1000m", "memory": "10G", "ephemeral-storage": "10G"},
    ),
    # Specifies path to kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # If true, the content of /airflow/xcom/return.json from container will
    # also be pushed to an XCom when the container ends.
    do_xcom_push=False,
    # List of Volume objects to pass to the Pod.
    volumes=[],
    # List of VolumeMount objects to pass to the Pod.
    volume_mounts=[],
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
    # Affinity determines which nodes the Pod can run on based on the
    # config. For more information see:
    # https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
    # Pod affinity with the KubernetesPodOperator
    # is not supported with Composer 2
    # instead, create a cluster and use the GKEStartPodOperator
    # https://cloud.google.com/composer/docs/using-gke-operator
    affinity={},
)

使用 Jinja 範本

Airflow 支援 DAG 中的 Jinja 範本。

您必須使用運算子宣告必要的 Airflow 參數 (task_id、name 和 image)。如下列範例所示，您可以使用 Jinja 為所有其他參數建立範本，包括 cmds、arguments、env_vars 和 config_file。

範例中的 env_vars 參數是從名為 my_value 的 Airflow 變數設定。範例 DAG 會從 Airflow 的 vars 範本變數取得值。Airflow 具有更多變數，可存取不同類型的資訊。舉例來說，您可以使用 conf 範本變數存取 Airflow 設定選項的值。如要進一步瞭解 Airflow 中可用的變數清單，請參閱 Airflow 說明文件中的「範本參考資料」。

在不變更 DAG 或建立 env_vars 變數的情況下，範例中的 ex-kube-templates 工作會失敗，因為變數不存在。在 Airflow 使用者介面或使用 Google Cloud CLI 建立這個變數：

Airflow UI

1. 前往 Airflow UI。

2. 在工具列中，依序選取「管理」>「變數」。

3. 在「List Variable」(清單變數) 頁面中，按一下「Add a new record」(新增記錄)。

4. 在「新增變數」頁面中輸入下列資訊：

   • 按鍵：my_value
   • Val：example_value

5. 按一下 [儲存]。

gcloud

輸入下列指令：

gcloud composer environments run ENVIRONMENT \
    --location LOCATION \
    variables set -- \
    my_value example_value

取代：

• 將 ENVIRONMENT 替換為環境的名稱。
• LOCATION 改成環境所在的地區。

以下範例說明如何搭配 KubernetesPodOperator 使用 Jinja 範本：

kubernetes_template_ex = KubernetesPodOperator(
    task_id="ex-kube-templates",
    name="ex-kube-templates",
    namespace="composer-user-workloads",
    image="bash",
    # All parameters below can be templated with Jinja. For more information
    # and the list of variables available in Airflow, see
    # the Airflow templates reference:
    # https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["echo"],
    # DS in Jinja is the execution date as YYYY-MM-DD, this Docker image
    # will echo the execution date. Arguments to the entrypoint. The Docker
    # image's CMD is used if this is not provided. The arguments parameter
    # is templated.
    arguments=["{{ ds }}"],
    # The var template variable allows you to access variables defined in
    # Airflow UI. In this case we are getting the value of my_value and
    # setting the environment variable `MY_VALUE`. The pod will fail if
    # `my_value` is not set in the Airflow UI. The env_vars parameter
    # is templated.
    env_vars={"MY_VALUE": "{{ var.value.my_value }}"},
    # Specifies path to Kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
)

使用 Kubernetes Secret 和 ConfigMap

Kubernetes Secret 是包含機密資料的物件。Kubernetes ConfigMap 是一個物件，包含鍵/值組合形式的非機密資料。

在 Cloud Composer 3 中，您可以使用 Google Cloud CLI、API 或 Terraform 建立 Secret 和 ConfigMap，然後從 KubernetesPodOperator 存取這些項目：

• 使用 Google Cloud CLI 和 API 時，您需要提供 YAML 設定檔。
• 使用 Terraform 時，您可以在 Terraform 設定檔中將 Secret 和 ConfigMap 定義為個別資源。

關於 YAML 設定檔

使用 Google Cloud CLI 和 API 建立 Kubernetes Secret 或 ConfigMap 時，您會提供 YAML 格式的檔案。這個檔案的格式必須與 Kubernetes Secret 和 ConfigMap 相同。Kubernetes 說明文件提供許多 ConfigMap 和 Secret 的程式碼範例。如要開始使用，請參閱「使用 Secret 安全地發布憑證」頁面和 ConfigMaps。

與 Kubernetes Secret 相同，在 Secret 中定義值時，請使用 base64 表示法。

注意： Base64 編碼的值不會以任何方式加密，只會經過模糊處理。可以從 Base64 表示法還原原始值。

如要編碼值，可以使用下列指令 (這是取得 Base64 編碼值的其中一種方式)：

echo "postgresql+psycopg2://root:example-password@127.0.0.1:3306/example-db" -n | base64

輸出：

cG9zdGdyZXNxbCtwc3ljb3BnMjovL3Jvb3Q6ZXhhbXBsZS1wYXNzd29yZEAxMjcuMC4wLjE6MzMwNi9leGFtcGxlLWRiIC1uCg==

本指南稍後的範例會使用下列兩個 YAML 檔案。 Kubernetes Secret 的 YAML 設定檔範例：

apiVersion: v1
kind: Secret
metadata:
  name: airflow-secrets
data:
  sql_alchemy_conn: cG9zdGdyZXNxbCtwc3ljb3BnMjovL3Jvb3Q6ZXhhbXBsZS1wYXNzd29yZEAxMjcuMC4wLjE6MzMwNi9leGFtcGxlLWRiIC1uCg==

另一個範例，說明如何納入檔案。與上一個範例相同，請先編碼檔案內容 (cat ./key.json | base64)，然後在 YAML 檔案中提供這個值：

apiVersion: v1
kind: Secret
metadata:
  name: service-account
data:
  service-account.json: |
    ewogICJ0eXBl...mdzZXJ2aWNlYWNjb3VudC5jb20iCn0K

ConfigMap 的 YAML 設定檔範例。您不需要在 ConfigMap 中使用 base64 表示法：

apiVersion: v1
kind: ConfigMap
metadata:
  name: example-configmap
data:
  example_key: example_value

管理 Kubernetes Secret

gcloud

建立 Secret

如要建立 Kubernetes Secret，請執行下列指令：

gcloud beta composer environments user-workloads-secrets create \
  --environment ENVIRONMENT_NAME \
  --location LOCATION \
  --secret-file-path SECRET_FILE

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。
• SECRET_FILE：包含 Secret 設定的本機 YAML 檔案路徑。

範例：

gcloud beta composer environments user-workloads-secrets create \
  --environment example-environment \
  --location us-central1 \
  --secret-file-path ./secrets/example-secret.yaml

更新 Secret

如要更新 Kubernetes Secret，請執行下列指令。系統會從指定的 YAML 檔案取得密鑰名稱，並取代密鑰內容。

gcloud beta composer environments user-workloads-secrets update \
  --environment ENVIRONMENT_NAME \
  --location LOCATION \
  --secret-file-path SECRET_FILE

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。
• SECRET_FILE：本機 YAML 檔案的路徑，其中包含 Secret 的設定。在這個檔案的 metadata > name 欄位中，指定密鑰名稱。

列出密鑰

如要取得環境的 Secret 清單及其欄位，請執行下列指令。輸出內容中的鍵值會替換為星號。

gcloud beta composer environments user-workloads-secrets list \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

取得密鑰詳細資料

如要取得 Secret 的詳細資訊，請執行下列指令。輸出內容中的鍵值會替換成星號。

gcloud beta composer environments user-workloads-secrets describe \
  SECRET_NAME \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

更改下列內容：

• SECRET_NAME：密鑰的名稱，如 YAML 檔案中 metadata > name 欄位所定義，其中包含密鑰的設定。
• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

刪除密鑰

如要刪除 Secret，請執行下列指令：

gcloud beta composer environments user-workloads-secrets delete \
  SECRET_NAME \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

• SECRET_NAME：密鑰的名稱，如 YAML 檔案中 metadata > name 欄位所定義，其中包含密鑰的設定。
• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

API

建立 Secret

1. 建立 environments.userWorkloadsSecrets.create API 要求。

2. 在這個要求中：

   1. 在要求主體的 name 欄位中，指定新密鑰的 URI。
   2. 在要求主體的 data 欄位中，指定 Secret 的金鑰和 base64 編碼值。

範例：

// POST https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsSecrets

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment/userWorkloadsSecrets/example-secret",
  "data": {
    "example": "ZXhhbXBsZV92YWx1ZSAtbgo="
  }
}

更新 Secret

1. 建立 environments.userWorkloadsSecrets.update API 要求。

2. 在這個要求中：

   1. 在要求主體的 name 欄位中，指定 Secret 的 URI。
   2. 在要求主體的 data 欄位中，指定 Secret 的金鑰和 base64 編碼值。系統會替換這些值。

範例：

// PUT https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsSecrets/example-secret

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment/userWorkloadsSecrets/example-secret",
  "data": {
    "example": "ZXhhbXBsZV92YWx1ZSAtbgo=",
    "another-example": "YW5vdGhlcl9leGFtcGxlX3ZhbHVlIC1uCg=="
  }
}

列出密鑰

建立 environments.userWorkloadsSecrets.list API 要求。輸出中的鍵值會替換成星號。您可以在這項要求中使用分頁功能，詳情請參閱要求參考資料。

範例：

// GET https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsSecrets

取得密鑰詳細資料

建立 environments.userWorkloadsSecrets.get API 要求。輸出中的鍵值會替換成星號。

範例：

// GET https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsSecrets/example-secret

刪除密鑰

建立 environments.userWorkloadsSecrets.delete API 要求。

範例：

// DELETE https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsSecrets/example-secret

Terraform

google_composer_user_workloads_secret 資源會定義 Kubernetes Secret，並在 data 區塊中定義鍵和值。

resource "google_composer_user_workloads_secret" "example_secret" {
  provider = google-beta
  environment = google_composer_environment.ENVIRONMENT_RESOURCE_NAME.name
  name = "SECRET_NAME"
  region = "LOCATION"

  data = {
    KEY_NAME: "KEY_VALUE"
  }
}

• ENVIRONMENT_RESOURCE_NAME：環境資源的名稱，其中包含 Terraform 中的環境定義。這個資源也會指定實際環境的名稱。
• LOCATION：環境所在的區域。
• SECRET_NAME：密鑰名稱。
• KEY_NAME：這個 Secret 的一或多個金鑰。
• KEY_VALUE：金鑰的 base64 編碼值。您可以使用 base64encode 函式編碼值 (請參閱範例)。

本指南稍後會使用下列兩個 Kubernetes Secret 範例。

resource "google_composer_user_workloads_secret" "example_secret" {
  provider = google-beta

  name = "airflow-secrets"

  environment = google_composer_environment.example_environment.name
  region = "us-central1"

  data = {
    sql_alchemy_conn: base64encode("postgresql+psycopg2://root:example-password@127.0.0.1:3306/example-db")
  }
}

另一個範例，說明如何納入檔案。您可以使用 file 函式將檔案內容讀取為字串，然後進行 Base64 編碼：

resource "google_composer_user_workloads_secret" "service_account_secret" {
  provider = google-beta

  name = "service-account"

  environment = google_composer_environment.example_environment.name
  region = "us-central1"

  data = {
    "service-account.json": base64encode(file("./key.json"))
  }
}

在 DAG 中使用 Kubernetes Secret

這個範例說明 Kubernetes Secret 的兩種用法：做為環境變數，以及做為 Pod 掛接的磁碟區。

第一個密鑰 airflow-secrets 會設為名為 SQL_CONN 的 Kubernetes 環境變數 (而非 Airflow 或 Cloud Composer 環境變數)。

第二個 Secret service-account 會將含有服務帳戶權杖的檔案 service-account.json 掛載至 /var/secrets/google。

Secret 物件如下所示：

secret_env = Secret(
    # Expose the secret as environment variable.
    deploy_type="env",
    # The name of the environment variable, since deploy_type is `env` rather
    # than `volume`.
    deploy_target="SQL_CONN",
    # Name of the Kubernetes Secret
    secret="airflow-secrets",
    # Key of a secret stored in this Secret object
    key="sql_alchemy_conn",
)
secret_volume = Secret(
    deploy_type="volume",
    # Path where we mount the secret as volume
    deploy_target="/var/secrets/google",
    # Name of Kubernetes Secret
    secret="service-account",
    # Key in the form of service account file name
    key="service-account.json",
)

第一個 Kubernetes Secret 的名稱是在 secret_env 變數中定義。這個 Secret 名稱為 airflow-secrets。deploy_type 參數會指定必須以環境變數形式公開。環境變數的名稱為 SQL_CONN，如 deploy_target 參數所指定。最後，SQL_CONN 環境變數的值會設為 sql_alchemy_conn 鍵的值。

第二個 Kubernetes Secret 的名稱定義於 secret_volume 變數中。這個 Secret 名稱為 service-account。如 deploy_type 參數所指定，這會公開為磁碟區。要掛接的檔案路徑 deploy_target 為 /var/secrets/google。最後，儲存在 deploy_target 中的 Secret 的 key 為 service-account.json。

運算子設定如下所示：

kubernetes_secret_vars_ex = KubernetesPodOperator(
    task_id="ex-kube-secrets",
    name="ex-kube-secrets",
    namespace="composer-user-workloads",
    image="gcr.io/gcp-runtimes/ubuntu_20_0_4",
    startup_timeout_seconds=300,
    # The secrets to pass to Pod, the Pod will fail to create if the
    # secrets you specify in a Secret object do not exist in Kubernetes.
    secrets=[secret_env, secret_volume],
    # Entrypoint of the container, if not specified the Docker container's
    # entrypoint is used. The cmds parameter is templated.
    cmds=["echo"],
    # env_vars allows you to specify environment variables for your
    # container to use. The env_vars parameter is templated.
    env_vars={
        "EXAMPLE_VAR": "/example/value",
        "GOOGLE_APPLICATION_CREDENTIALS": "/var/secrets/google/service-account.json",
    },
    # Specifies path to kubernetes config. The config_file is templated.
    config_file="/home/airflow/composer_kube_config",
    # Identifier of connection that should be used
    kubernetes_conn_id="kubernetes_default",
)

管理 Kubernetes ConfigMap

gcloud

建立 ConfigMap

如要建立 ConfigMap，請執行下列指令：

gcloud beta composer environments user-workloads-config-maps create \
  --environment ENVIRONMENT_NAME \
  --location LOCATION \
  --config-map-file-path CONFIG_MAP_FILE

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。
• CONFIG_MAP_FILE：包含 ConfigMap 設定的本機 YAML 檔案路徑。

範例：

gcloud beta composer environments user-workloads-config-maps create \
  --environment example-environment \
  --location us-central1 \
  --config-map-file-path ./configs/example-configmap.yaml

更新 ConfigMap

如要更新 ConfigMap，請執行下列指令。ConfigMap 的名稱會取自指定的 YAML 檔案，且 ConfigMap 的內容會遭到取代。

gcloud beta composer environments user-workloads-config-maps update \
  --environment ENVIRONMENT_NAME \
  --location LOCATION \
  --config-map-file-path CONFIG_MAP_FILE

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。
• CONFIG_MAP_FILE：包含 ConfigMap 設定的本機 YAML 檔案路徑。在這個檔案的 metadata > name 欄位中，指定 ConfigMap 的名稱。

列出 ConfigMap

如要取得環境的 ConfigMap 清單及其欄位，請執行下列指令。輸出內容中的鍵值會照原樣顯示。

gcloud beta composer environments user-workloads-config-maps list \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

更改下列內容：

• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

取得 ConfigMap 的詳細資料

如要取得 ConfigMap 的詳細資訊，請執行下列指令。輸出內容中的鍵/值會照原樣顯示。

gcloud beta composer environments user-workloads-config-maps describe \
  CONFIG_MAP_NAME \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

更改下列內容：

• CONFIG_MAP_NAME：ConfigMap 的名稱，如 YAML 檔案中 metadata > name 欄位所定義，其中包含 ConfigMap 的設定。
• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

刪除 ConfigMap

如要刪除 ConfigMap，請執行下列指令：

gcloud beta composer environments user-workloads-config-maps delete \
  CONFIG_MAP_NAME \
  --environment ENVIRONMENT_NAME \
  --location LOCATION

• CONFIG_MAP_NAME：ConfigMap 的名稱，如 YAML 檔案中 metadata > name 欄位所定義，其中包含 ConfigMap 的設定。
• ENVIRONMENT_NAME：環境名稱。
• LOCATION：環境所在的區域。

API

建立 ConfigMap

1. 建立 API 要求。environments.userWorkloadsConfigMaps.create

2. 在這個要求中：

   1. 在要求主體的 name 欄位中，指定新 ConfigMap 的 URI。
   2. 在要求主體的 data 欄位中，指定 ConfigMap 的鍵和值。

範例：

// POST https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsConfigMaps

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment/userWorkloadsConfigMaps/example-configmap",
  "data": {
    "example_key": "example_value"
  }
}

更新 ConfigMap

1. 建立 API 要求。environments.userWorkloadsConfigMaps.update

2. 在這個要求中：

   1. 在要求主體的 name 欄位中，指定 ConfigMap 的 URI。
   2. 在要求主體的 data 欄位中，指定 ConfigMap 的鍵和值。系統會替換這些值。

範例：

// PUT https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsConfigMaps/example-configmap

{
  "name": "projects/example-project/locations/us-central1/environments/example-environment/userWorkloadsConfigMaps/example-configmap",
  "data": {
    "example_key": "example_value",
    "another_key": "another_value"
  }
}

列出 ConfigMap

建立 environments.userWorkloadsConfigMaps.list API 要求。輸出內容中的鍵值會照原樣顯示。您可以在這項要求中使用分頁功能，詳情請參閱要求參考資料。

範例：

// GET https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsConfigMaps

取得 ConfigMap 的詳細資料

建立 environments.userWorkloadsConfigMaps.get API 要求。輸出內容中的鍵值會照原樣顯示。

範例：

// GET https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsConfigMaps/example-configmap

刪除 ConfigMap

建立 environments.userWorkloadsConfigMaps.delete API 要求。

範例：

// DELETE https://composer.googleapis.com/v1beta1/projects/example-project/
// locations/us-central1/environments/example-environment/userWorkloadsConfigMaps/example-configmap

Terraform

google_composer_user_workloads_config_map 資源會定義 ConfigMap，並在 data 區塊中定義鍵和值。

resource "google_composer_user_workloads_config_map" "example_config_map" {
  provider = google-beta
  environment = google_composer_environment.ENVIRONMENT_RESOURCE_NAME.name
  name = "CONFIG_MAP_NAME"
  region = "LOCATION"

  data = {
    KEY_NAME: "KEY_VALUE"
  }
}

• ENVIRONMENT_RESOURCE_NAME：環境資源的名稱，其中包含 Terraform 中的環境定義。這個資源也會指定實際環境的名稱。
• LOCATION：環境所在的區域。
• CONFIG_MAP_NAME：ConfigMap 的名稱。
• KEY_NAME：這個 ConfigMap 的一或多個鍵。
• KEY_VALUE：鍵的值。

範例：

resource "google_composer_user_workloads_config_map" "example_config_map" {
  provider = google-beta

  name = "example-config-map"

  environment = google_composer_environment.example_environment.name
  region = "us-central1"

  data = {
    "example_key": "example_value"
  }
}

在 DAG 中使用 ConfigMap

這個範例說明如何在 DAG 中使用 ConfigMap。

在下列範例中，ConfigMap 會在 configmaps 參數中傳遞。這個 ConfigMap 的所有鍵都可做為環境變數：

import datetime

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

with models.DAG(
    dag_id="composer_kubernetes_pod_configmap",
    schedule_interval=None,
    start_date=datetime.datetime(2024, 1, 1),
) as dag:

  KubernetesPodOperator(
    task_id='kpo_configmap_env_vars',
    image='busybox:1.28',
    cmds=['sh'],
    arguments=[
        '-c',
        'echo "Value: $example_key"',
    ],
    configmaps=["example-configmap"],
    config_file="/home/airflow/composer_kube_config",
  )

以下範例說明如何將 ConfigMap 掛接為磁碟區：

import datetime

from airflow import models
from kubernetes.client import models as k8s
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator

volume_mount = k8s.V1VolumeMount(name='confmap-example',
  mount_path='/config',
  sub_path=None,
  read_only=False)

volume = k8s.V1Volume(name='confmap-example',
  config_map=k8s.V1ConfigMapVolumeSource(name='example-configmap'))

with models.DAG(
    dag_id="composer_kubernetes_pod_configmap",
    schedule_interval=None,
    start_date=datetime.datetime(2024, 1, 1),
) as dag:

  KubernetesPodOperator(
    task_id='kpo_configmap_volume_mount',
    image='busybox:1.28',
    cmds=['sh'],
    arguments=[
        '-c',
        'ls /config'
    ],
    volumes=[volume],
    volume_mounts=[volume_mount],
    configmaps=["example-configmap"],
    config_file="/home/airflow/composer_kube_config",
  )

CNCF Kubernetes 供應商相關資訊

KubernetesPodOperator 會在 apache-airflow-providers-cncf-kubernetes provider 中實作。

如需 CNCF Kubernetes 供應商的詳細版本資訊，請參閱 CNCF Kubernetes 供應商網站。

資源需求

Cloud Composer 3 支援下列資源需求值。如需使用資源需求的範例，請參閱其他設定。

資源	下限	上限	步驟
CPU	0.25	32	間距值：0.25、0.5、1、2、4、6、8、10、...、32。系統會將要求的值無條件進位至最接近的支援步驟值 (例如 5 進位至 6)。
記憶體	2G (GB)	128G (GB)	間距值：2、3、4、5、...、128。系統會將要求的值無條件進位至最接近的支援步驟值 (例如 3.5G 進位至 4G)。
儲存空間	-	100G (GB)	任何值。如果要求超過 100 GB，系統只會提供 100 GB。

如要進一步瞭解 Kubernetes 中的資源單位，請參閱「Kubernetes 中的資源單位」。

疑難排解

本節提供建議，協助您排解常見的 KubernetesPodOperator 問題：

查看記錄

排解問題時，您可以依下列順序檢查記錄：

1. Airflow 工作記錄：

   1. 前往 Google Cloud 控制台的「Environments」頁面。

      前往「環境」

   2. 在環境清單中，按一下環境名稱。 「環境詳細資料」頁面隨即開啟。

   3. 前往「DAG」分頁。

   4. 按一下 DAG 名稱，然後按一下 DAG 執行作業，即可查看詳細資料和記錄。

2. Airflow 排程器記錄：

   1. 前往「環境詳細資料」頁面。

   2. 前往「記錄」分頁。

   3. 檢查 Airflow 排程器記錄。

3. 使用者工作負載記錄：

   1. 前往「環境詳細資料」頁面。

   2. 前往「監控」分頁。

   3. 選取「使用者工作負載」。

   4. 檢查已執行的工作負載清單。您可以查看每個工作負載的記錄和資源用量資訊。

非零的傳回代碼

使用 KubernetesPodOperator (和 GKEStartPodOperator) 時，容器進入點的回傳代碼會決定工作是否視為成功。非零回傳代碼表示失敗。

常見模式是將殼層指令碼做為容器進入點執行，以便在容器內將多項作業歸為一組。

如果您要編寫這類指令碼，建議在指令碼頂端加入 set -e 指令，這樣一來，指令碼中失敗的指令就會終止指令碼，並將失敗傳播至 Airflow 工作執行個體。

Pod 逾時

KubernetesPodOperator 的預設逾時時間為 120 秒，這可能會導致系統在下載較大的映像檔前逾時。建立 KubernetesPodOperator 時，您可以變更 startup_timeout_seconds 參數來延長逾時時間。

Pod 超時時，您可以在 Airflow UI 中查看特定工作記錄。例如：

Executing <Task(KubernetesPodOperator): ex-all-configs> on 2018-07-23 19:06:58.133811
Running: ['bash', '-c', u'airflow run kubernetes-pod-example ex-all-configs 2018-07-23T19:06:58.133811 --job_id 726 --raw -sd DAGS_FOLDER/kubernetes_pod_operator_sample.py']
Event: pod-name-9a8e9d06 had an event of type Pending
...
...
Event: pod-name-9a8e9d06 had an event of type Pending
Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 27, in <module>
    args.func(args)
  File "/usr/local/lib/python2.7/site-packages/airflow/bin/cli.py", line 392, in run
    pool=args.pool,
  File "/usr/local/lib/python2.7/site-packages/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 1492, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/python2.7/site-packages/airflow/contrib/operators/kubernetes_pod_operator.py", line 123, in execute
    raise AirflowException('Pod Launching failed: {error}'.format(error=ex))
airflow.exceptions.AirflowException: Pod Launching failed: Pod took too long to start

如果 Cloud Composer 服務帳戶缺少執行目前工作所需的 IAM 權限，也可能發生 Pod 超時問題。如要驗證這點，請使用 GKE 資訊主頁查看 Pod 層級的錯誤，找出特定工作負載的記錄，或使用 Cloud Logging。

執行大量工作時，KubernetesPodOperator 工作會失敗

如果環境同時執行大量 KubernetesPodOperator 或 KubernetesExecutor 工作，Cloud Composer 3 會等到部分現有工作完成後，才會接受新工作。

如要進一步瞭解如何排解這個問題，請參閱「排解 KubernetesExecutor 工作問題」。

後續步驟

• 使用 GKE 運算子
• Kubernetes 基礎知識