El 15 de septiembre del 2026, todos los entornos de Cloud Composer 1 y Cloud Composer 2 versión 2.0.x alcanzarán el final de su ciclo de vida previsto y no podrás usarlos. Te recomendamos que planifiques la migración a Cloud Composer 3.

Esta página se ha traducido con Cloud Translation API.

Usar KubernetesPodOperador

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

En esta página se describe cómo usar KubernetesPodOperator para desplegar pods de Kubernetes desde Cloud Composer en el clúster de Google Kubernetes Engine que forma parte de tu entorno de Cloud Composer.

KubernetesPodOperator inicia pods de Kubernetes en el clúster de tu entorno. En comparación, los operadores de Google Kubernetes Engine ejecutan pods de Kubernetes en un clúster específico, que puede ser un clúster independiente que no esté relacionado con tu entorno. También puedes crear y eliminar clústeres con los operadores de Google Kubernetes Engine.

KubernetesPodOperator es una buena opción si necesitas lo siguiente:

Dependencias de Python personalizadas que no están disponibles en el repositorio público de PyPI.
Dependencias binarias que no están disponibles en la imagen de trabajador de Cloud Composer.

Antes de empezar

Consulta la siguiente lista de diferencias entre KubernetesPodOperator en Cloud Composer 3 y Cloud Composer 2, y asegúrate de que tus DAGs sean compatibles:

No se pueden crear espacios de nombres personalizados en Cloud Composer 3. Los pods siempre se ejecutan en el espacio de nombres composer-user-workloads, aunque se especifique otro. Los pods de este espacio de nombres tienen acceso a los recursos y a la red VPC de tu proyecto (si está habilitada) sin necesidad de configuración adicional.
No es posible ejecutar varios contenedores sidecar adicionales en Cloud Composer 3. Puedes ejecutar un solo contenedor auxiliar si se llama airflow-xcom-sidecar.
Los secretos y los ConfigMaps de Kubernetes no se pueden crear con la API de Kubernetes. En su lugar, Cloud Composer proporciona comandos de Google Cloud CLI, recursos de Terraform y la API de Cloud Composer para gestionar los secretos y los ConfigMaps de Kubernetes. Para obtener más información, consulta Usar secretos y ConfigMaps de Kubernetes.
No es posible implementar cargas de trabajo personalizadas en Cloud Composer 3. Solo se pueden modificar los secretos y los ConfigMaps de Kubernetes, pero no se pueden hacer otros cambios en la configuración.
Los requisitos de recursos (CPU, memoria y almacenamiento) deben especificarse con valores admitidos.
Al igual que en Cloud Composer 2, la configuración de afinidad de pods no está disponible. Si quieres usar la afinidad de pods, usa los operadores de GKE para iniciar pods en otro clúster.

Acerca de KubernetesPodOperator en Cloud Composer 3

En esta sección se describe cómo funciona KubernetesPodOperator en Cloud Composer 3.

Uso de recursos

En Cloud Composer 3, el clúster de tu entorno se escala automáticamente. Las cargas de trabajo adicionales que ejecutes con KubernetesPodOperator se escalan de forma independiente de tu entorno. Tu entorno no se ve afectado por el aumento de la demanda de recursos, pero el clúster de tu entorno se amplía y se reduce en función de la demanda de recursos.

Los precios de las cargas de trabajo adicionales que ejecutes en el clúster de tu entorno se corresponden con el modelo de precios de Cloud Composer 3 y utilizan los SKUs de Cloud Composer 3.

Cloud Composer 3 usa clústeres de Autopilot, que introducen el concepto de clases de computación:

Cloud Composer solo admite la clase de computación general-purpose.
De forma predeterminada, si no se selecciona ninguna clase, se asume la clase general-purpose al crear pods con KubernetesPodOperator.
Cada clase está asociada a propiedades y límites de recursos específicos. Puedes consultar más información en la documentación de Autopilot. Por ejemplo, los pods que se ejecutan en la clase general-purpose pueden usar hasta 110 GiB de memoria.

Acceso a los recursos del proyecto

En Cloud Composer 3, el clúster de tu entorno se encuentra en el proyecto de inquilino y no se puede configurar. Los pods se ejecutan en el clúster del entorno, en un espacio de nombres aislado.

En Cloud Composer 3, los pods siempre se ejecutan en el espacio de nombres composer-user-workloads, aunque se especifique otro. Los pods de este espacio de nombres pueden acceder a los recursos de tu proyecto y a tu red de VPC (si está habilitada) sin necesidad de configuración adicional. Google Cloud La cuenta de servicio de tu entorno se usa para acceder a estos recursos. No es posible especificar otra cuenta de servicio.

Configuración mínima

Para crear un KubernetesPodOperator, solo se necesitan los parámetros name, image y task_id del pod que se va a usar. El /home/airflow/composer_kube_config contiene las credenciales para autenticarte en 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",
)

Configuración adicional

En este ejemplo se muestran parámetros adicionales que puedes configurar en KubernetesPodOperator.

Consulta los siguientes recursos para obtener más información:

Para obtener información sobre cómo usar los secretos y los ConfigMaps de Kubernetes, consulta Usar secretos y ConfigMaps de Kubernetes.
Para obtener información sobre cómo usar plantillas Jinja con KubernetesPodOperator, consulta Usar plantillas Jinja.
Para obtener información sobre los valores admitidos de los requisitos de recursos (CPU, memoria y almacenamiento), consulta Requisitos de recursos.
Para obtener información sobre los parámetros de KubernetesPodOperator, consulta la referencia del operador en la documentación de 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={},
)

Usar plantillas Jinja

Airflow admite plantillas Jinja en los DAGs.

Debe declarar los parámetros de Airflow obligatorios (task_id, name y image) con el operador. Como se muestra en el siguiente ejemplo, puedes usar Jinja para crear plantillas de todos los demás parámetros, incluidos cmds, arguments, env_vars y config_file.

El parámetro env_vars del ejemplo se define a partir de una variable de Airflow llamada my_value. El DAG de ejemplo obtiene su valor de la variable de plantilla vars de Airflow. Airflow tiene más variables que proporcionan acceso a diferentes tipos de información. Por ejemplo, puedes usar la variable de plantilla conf para acceder a los valores de las opciones de configuración de Airflow. Para obtener más información y la lista de variables disponibles en Airflow, consulta la referencia de plantillas en la documentación de Airflow.

Sin cambiar el DAG ni crear la variable env_vars, la tarea ex-kube-templates del ejemplo falla porque la variable no existe. Crea esta variable en la interfaz de usuario de Airflow o con Google Cloud CLI:

Interfaz de usuario de Airflow

Ve a la interfaz de Airflow.
En la barra de herramientas, seleccione Administrar > Variables.
En la página List Variable (Variable de lista), haz clic en Add a new record (Añadir un nuevo registro).
En la página Añadir variable, introduce la siguiente información:
- Tecla:my_value
- Valor: example_value
Haz clic en Guardar.

gcloud

Introduce el siguiente comando:

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

Sustituye:

ENVIRONMENT con el nombre del entorno.
LOCATION con la región en la que se encuentra el entorno.

En el siguiente ejemplo se muestra cómo usar plantillas Jinja con KubernetesPodOperator:

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",
)

Usar secretos y ConfigMaps de Kubernetes

Un secreto de Kubernetes es un objeto que contiene datos sensibles. Un ConfigMap de Kubernetes es un objeto que contiene datos no confidenciales en pares clave-valor.

En Cloud Composer 3, puedes crear secretos y ConfigMaps con la CLI de Google Cloud, la API o Terraform, y, a continuación, acceder a ellos desde KubernetesPodOperator:

Con la CLI y la API de Google Cloud, debes proporcionar un archivo de configuración YAML.
Con Terraform, puedes definir secretos y ConfigMaps como recursos independientes en archivos de configuración de Terraform.

Acerca de los archivos de configuración YAML

Cuando creas un Secret o un ConfigMap de Kubernetes con la CLI de Google Cloud y la API, proporcionas un archivo en formato YAML. Este archivo debe tener el mismo formato que los secretos y los ConfigMaps de Kubernetes. La documentación de Kubernetes proporciona muchos ejemplos de código de ConfigMaps y Secrets. Para empezar, puedes consultar la página Distribuir credenciales de forma segura mediante secretos y ConfigMaps.

Al igual que en los secretos de Kubernetes, usa la representación en base64 cuando definas valores en los secretos.

Para codificar un valor, puedes usar el siguiente comando (esta es una de las muchas formas de obtener un valor codificado en base64):

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

Resultado:

cG9zdGdyZXNxbCtwc3ljb3BnMjovL3Jvb3Q6ZXhhbXBsZS1wYXNzd29yZEAxMjcuMC4wLjE6MzMwNi9leGFtcGxlLWRiIC1uCg==

En los ejemplos que se muestran más adelante en esta guía se utilizan los dos archivos YAML que se indican a continuación. Ejemplo de archivo de configuración YAML de un secreto de Kubernetes:

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

Otro ejemplo que muestra cómo incluir archivos. Al igual que en el ejemplo anterior, primero codifica el contenido de un archivo (cat ./key.json | base64) y, a continuación, proporciona este valor en el archivo YAML:

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

Ejemplo de archivo de configuración YAML de un ConfigMap. No es necesario que utilices la representación base64 en ConfigMaps:

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

Gestionar secretos de Kubernetes

gcloud

Crear un secreto

Para crear un secreto de Kubernetes, ejecuta el siguiente comando:

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.
SECRET_FILE: ruta a un archivo YAML local que contiene la configuración del secreto.

Ejemplo:

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

Actualizar un secreto

Para actualizar un secreto de Kubernetes, ejecuta el siguiente comando. El nombre del secreto se tomará del archivo YAML especificado y se sustituirá el contenido del secreto.

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.
SECRET_FILE: ruta a un archivo YAML local que contiene la configuración del secreto. Especifica el nombre del secreto en el campo metadata > name de este archivo.

List Secrets

Para obtener una lista de los secretos y sus campos de un entorno, ejecuta el siguiente comando. Los valores de las claves de la salida se sustituirán por asteriscos.

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

Obtener los detalles de Secret

Para obtener información detallada sobre un secreto, ejecuta el siguiente comando. Los valores de las claves de la salida se sustituirán por asteriscos.

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

Haz los cambios siguientes:

SECRET_NAME: el nombre del secreto, tal como se definió en el campo metadata > name del archivo YAML con la configuración del secreto.
ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

Eliminar un secreto

Para eliminar un secreto, ejecuta el siguiente comando:

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

SECRET_NAME: el nombre del secreto, tal como se definió en el campo metadata > name del archivo YAML con la configuración del secreto.
ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

API

Crear un secreto

Crea una solicitud de API environments.userWorkloadsSecrets.create.
En esta solicitud:
1. En el cuerpo de la solicitud, en el campo name, especifica el URI del nuevo secreto.
2. En el cuerpo de la solicitud, en el campo data, especifica las claves y los valores codificados en Base64 del secreto.

Ejemplo:

// 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="
  }
}

Actualizar un secreto

Crea una solicitud de API environments.userWorkloadsSecrets.update.
En esta solicitud:
1. En el cuerpo de la solicitud, en el campo name, especifica el URI del secreto.
2. En el cuerpo de la solicitud, en el campo data, especifica las claves y los valores codificados en Base64 del secreto. Los valores se sustituirán.

Ejemplo:

// 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=="
  }
}

List Secrets

Crea una solicitud de API environments.userWorkloadsSecrets.list. Los valores de clave de la salida se sustituirán por asteriscos. Es posible usar la paginación con esta solicitud. Consulta la referencia de la solicitud para obtener más información.

Ejemplo:

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

Obtener los detalles de Secret

Crea una solicitud de API environments.userWorkloadsSecrets.get. Los valores de clave de la salida se sustituirán por asteriscos.

Ejemplo:

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

Eliminar un secreto

Crea una solicitud de API environments.userWorkloadsSecrets.delete.

Ejemplo:

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

Terraform

El recurso google_composer_user_workloads_secret define un secreto de Kubernetes, con claves y valores definidos en el bloque 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: el nombre del recurso del entorno, que contiene la definición del entorno en Terraform. El nombre del entorno real también se especifica en este recurso.
LOCATION: la región en la que se encuentra el entorno.
SECRET_NAME: el nombre del secreto.
KEY_NAME: una o varias claves de este secreto.
KEY_VALUE: valor codificado en base64 de la clave. Puede usar la función base64encode para codificar el valor (consulte el ejemplo).

Los dos ejemplos siguientes de secretos de Kubernetes se usan en ejemplos posteriores de esta guía.

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")
  }
}

Otro ejemplo que muestra cómo incluir archivos. Puedes usar la función file para leer el contenido del archivo como una cadena y, a continuación, codificarlo en 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"))
  }
}

Usar secretos de Kubernetes en tus DAGs

En este ejemplo se muestran dos formas de usar los secretos de Kubernetes: como variable de entorno y como volumen montado por el pod.

El primer secreto, airflow-secrets, se asigna a una variable de entorno de Kubernetes llamada SQL_CONN (en lugar de a una variable de entorno de Airflow o Cloud Composer).

El segundo secreto, service-account, monta service-account.json, un archivo con un token de cuenta de servicio, en /var/secrets/google.

Este es el aspecto de los objetos secretos:

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",
)

El nombre del primer secreto de Kubernetes se define en la variable secret_env. Este secreto se llama airflow-secrets. El parámetro deploy_type especifica que debe exponerse como una variable de entorno. El nombre de la variable de entorno es SQL_CONN, tal como se especifica en el parámetro deploy_target. Por último, el valor de la variable de entorno SQL_CONN se asigna al valor de la clave sql_alchemy_conn.

El nombre del segundo secreto de Kubernetes se define en la variable secret_volume. Este secreto se llama service-account. Se expone como un volumen, tal como se especifica en el parámetro deploy_type. La ruta del archivo que se va a montar, deploy_target, es /var/secrets/google. Por último, el key del secreto almacenado en deploy_target es service-account.json.

Este es el aspecto de la configuración del operador:

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",
)

Gestionar ConfigMaps de Kubernetes

gcloud

Crear un ConfigMap

Para crear un ConfigMap, ejecuta el siguiente comando:

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.
CONFIG_MAP_FILE: ruta a un archivo YAML local que contiene la configuración de ConfigMap.

Ejemplo:

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

Actualizar un ConfigMap

Para actualizar un ConfigMap, ejecuta el siguiente comando. El nombre de ConfigMaps se tomará del archivo YAML especificado y se sustituirá el contenido de ConfigMap.

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.
CONFIG_MAP_FILE: ruta a un archivo YAML local que contiene la configuración de ConfigMap. Especifica el nombre de ConfigMap en el campo metadata > name de este archivo.

Listar ConfigMaps

Para obtener una lista de los ConfigMaps y sus campos de un entorno, ejecuta el siguiente comando. Los valores clave de la salida se mostrarán tal cual.

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

Haz los cambios siguientes:

ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

Obtener los detalles de un ConfigMap

Para obtener información detallada sobre un ConfigMap, ejecuta el siguiente comando. Los valores de Key en el resultado se mostrarán tal cual.

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

Haz los cambios siguientes:

CONFIG_MAP_NAME: el nombre de ConfigMap, tal como se definió en el campo metadata > name del archivo YAML con la configuración de ConfigMap.
ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

Eliminar un ConfigMap

Para eliminar un ConfigMap, ejecuta el siguiente comando:

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

CONFIG_MAP_NAME: el nombre de ConfigMap, tal como se definió en el campo metadata > name del archivo YAML con la configuración de ConfigMap.
ENVIRONMENT_NAME: el nombre de tu entorno.
LOCATION: la región en la que se encuentra el entorno.

API

Crear un ConfigMap

Crea una solicitud de la API environments.userWorkloadsConfigMaps.create.
En esta solicitud:
1. En el cuerpo de la solicitud, en el campo name, especifica el URI del nuevo ConfigMap.
2. En el cuerpo de la solicitud, en el campo data, especifica las claves y los valores de ConfigMap.

Ejemplo:

// 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"
  }
}

Actualizar un ConfigMap

Crea una solicitud de la API environments.userWorkloadsConfigMaps.update.
En esta solicitud:
1. En el cuerpo de la solicitud, en el campo name, especifica el URI de ConfigMap.
2. En el cuerpo de la solicitud, en el campo data, especifica las claves y los valores de ConfigMap. Los valores se sustituirán.

Ejemplo:

// 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"
  }
}

Listar ConfigMaps

Crea una solicitud de API environments.userWorkloadsConfigMaps.list. Los valores de las claves de la salida se mostrarán tal cual. Es posible usar la paginación con esta solicitud. Consulta la referencia de la solicitud para obtener más información.

Ejemplo:

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

Obtener los detalles de un ConfigMap

Crea una solicitud de API environments.userWorkloadsConfigMaps.get. Los valores clave de la salida se mostrarán tal cual.

Ejemplo:

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

Eliminar un ConfigMap

Crea una solicitud de API environments.userWorkloadsConfigMaps.delete.

Ejemplo:

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

Terraform

El recurso google_composer_user_workloads_config_map define un ConfigMap con claves y valores definidos en el bloque 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: el nombre del recurso del entorno, que contiene la definición del entorno en Terraform. El nombre del entorno real también se especifica en este recurso.
LOCATION: la región en la que se encuentra el entorno.
CONFIG_MAP_NAME: nombre del ConfigMap.
KEY_NAME: una o varias claves de este ConfigMap.
KEY_VALUE: valor de la clave.

Ejemplo:

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"
  }
}

Usar ConfigMaps en tus DAGs

En este ejemplo se muestra cómo usar ConfigMaps en tus DAGs.

En el ejemplo siguiente, se transfiere un ConfigMap en el parámetro configmaps. Todas las claves de este ConfigMap están disponibles como variables de entorno:

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",
  )

En el siguiente ejemplo se muestra cómo montar un ConfigMap como un volumen:

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",
  )

Información sobre el proveedor de Kubernetes de la CNCF

KubernetesPodOperator se implementa en el proveedor apache-airflow-providers-cncf-kubernetes.

Para consultar las notas de la versión detalladas del proveedor de Kubernetes de CNCF, visita el sitio web del proveedor de Kubernetes de CNCF.

Requisitos de recursos

Cloud Composer 3 admite los siguientes valores para los requisitos de recursos. Para ver un ejemplo de cómo usar los requisitos de recursos, consulta Configuración adicional.

Recurso	Mínimo	Máximo	Paso
CPU	0,25	32	Valores de incremento: 0,25, 0,5, 1, 2, 4, 6, 8, 10, ..., 32. Los valores solicitados se redondean al valor de paso admitido más cercano (por ejemplo, de 5 a 6).
Memoria	2G (GB)	128 G (GB)	Valores del incremento: 2, 3, 4, 5, ..., 128. Los valores solicitados se redondean al valor de paso admitido más cercano (por ejemplo, 3,5 G se redondea a 4 G).
Almacenamiento	-	100G (GB)	Cualquier valor. Si se solicitan más de 100 GB, solo se proporcionarán 100 GB.

Para obtener más información sobre las unidades de recursos en Kubernetes, consulta Unidades de recursos en Kubernetes.

Solución de problemas

En esta sección se ofrecen consejos para solucionar problemas habituales de KubernetesPodOperator:

Ver registros

Cuando intentes solucionar problemas, puedes consultar los registros en el siguiente orden:

Registros de tareas de Airflow:
1. En la Google Cloud consola, ve a la página Entornos.
  
  Ir a Entornos
2. En la lista de entornos, haz clic en el nombre del entorno. Se abrirá la página Detalles del entorno.
3. Ve a la pestaña DAGs.
4. Haga clic en el nombre del DAG y, a continuación, en la ejecución del DAG para ver los detalles y los registros.
Registros del programador de Airflow:
1. Ve a la página Detalles del entorno.
2. Ve a la pestaña Registros.
3. Inspecciona los registros del programador de Airflow.
Registros de cargas de trabajo de usuarios:
1. Ve a la página Detalles del entorno.
2. Vaya a la pestaña Monitorización.
3. Selecciona Cargas de trabajo de usuario.
4. Consulta la lista de cargas de trabajo ejecutadas. Puedes ver los registros y la información de uso de recursos de cada carga de trabajo.

Códigos de retorno distintos de cero

Cuando se usa KubernetesPodOperator (y GKEStartPodOperator), el código de retorno del punto de entrada del contenedor determina si la tarea se considera completada correctamente o no. Los códigos de retorno distintos de cero indican un error.

Un patrón habitual es ejecutar un script de shell como punto de entrada del contenedor para agrupar varias operaciones en el contenedor.

Si vas a escribir una secuencia de comandos de este tipo, te recomendamos que incluyas el comando set -e en la parte superior para que, si se produce un error en un comando de la secuencia, esta se detenga y el error se propague a la instancia de tarea de Airflow.

Tiempos de espera de pods

El tiempo de espera predeterminado de KubernetesPodOperator es de 120 segundos, lo que puede provocar que se agote el tiempo de espera antes de que se descarguen imágenes más grandes. Puedes aumentar el tiempo de espera modificando el parámetro startup_timeout_seconds al crear el KubernetesPodOperator.

Cuando se agota el tiempo de espera de un pod, el registro específico de la tarea está disponible en la interfaz de usuario de Airflow. Por ejemplo:

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

Los tiempos de espera de los pods también pueden producirse cuando la cuenta de servicio de Cloud Composer no tiene los permisos de gestión de identidades y accesos necesarios para realizar la tarea en cuestión. Para verificarlo, consulta los errores a nivel de pod en los paneles de control de GKE para ver los registros de tu carga de trabajo específica o usa Cloud Logging.

Las tareas de KubernetesPodOperator fallan cuando se ejecuta un gran número de tareas

Cuando tu entorno ejecuta un gran número de tareas de KubernetesPodOperator o KubernetesExecutor al mismo tiempo, Cloud Composer 3 no acepta nuevas tareas hasta que se completen algunas de las tareas existentes.

Para obtener más información sobre cómo solucionar este problema, consulta Solucionar problemas de tareas de KubernetesExecutor.

Usar KubernetesPodOperador Organízate con las colecciones Guarda y clasifica el contenido según tus preferencias.

Antes de empezar

Acerca de KubernetesPodOperator en Cloud Composer 3

Uso de recursos

Acceso a los recursos del proyecto

Configuración mínima

Configuración adicional

Usar plantillas Jinja

Interfaz de usuario de Airflow

gcloud

Usar secretos y ConfigMaps de Kubernetes

Acerca de los archivos de configuración YAML

Gestionar secretos de Kubernetes

gcloud

API

Terraform

Usar secretos de Kubernetes en tus DAGs

Gestionar ConfigMaps de Kubernetes

gcloud

API

Terraform

Usar ConfigMaps en tus DAGs

Información sobre el proveedor de Kubernetes de la CNCF

Requisitos de recursos

Solución de problemas

Ver registros

Códigos de retorno distintos de cero

Tiempos de espera de pods

Las tareas de KubernetesPodOperator fallan cuando se ejecuta un gran número de tareas

Siguientes pasos

Usar KubernetesPodOperador