שימוש ב-KubernetesPodOperator

‫Managed Airflow (דור 3) | Managed Airflow (דור 2) | Managed Airflow (דור 1 מדור קודם)

בדף הזה מוסבר איך להשתמש ב-KubernetesPodOperator כדי לפרוס Kubernetes Pods מ-Managed Service for Apache Airflow אל אשכול Google Kubernetes Engine שמהווה חלק מהסביבה של Managed Service for Apache Airflow.

האופרטור KubernetesPodOperator מפעיל קבוצות Pod של Kubernetes באשכול שבסביבה שלכם. לעומת זאת, אופרטורים של Google Kubernetes Engine מריצים Pods של Kubernetes באשכול שצוין, שיכול להיות אשכול נפרד שלא קשור לסביבה שלכם. אפשר גם ליצור ולמחוק אשכולות באמצעות אופרטורים של Google Kubernetes Engine.

‫KubernetesPodOperator היא אפשרות טובה אם אתם צריכים:

• יחסי תלות מותאמים אישית של Python שלא זמינים דרך מאגר PyPI הציבורי.
• תלויות בינאריות שלא זמינות בתמונת ה-worker של Managed Airflow.

הערה: מידע נוסף על התקנת חבילות שמתארחות במאגר חבילות פרטי או על אפשרויות נוספות לניהול יחסי תלות זמין במאמר התקנת יחסי תלות ב-Python.

לפני שמתחילים

• אם משתמשים בגרסה 5.0.0 של CNCF Kubernetes Provider, צריך לפעול לפי ההוראות שמפורטות בקטע CNCF Kubernetes Provider.

• ההגדרה של שיוך Pod לא זמינה ב-Managed Airflow (דור 2). אם רוצים להשתמש בהצמדה של Pod, צריך להשתמש באופרטורים של GKE כדי להפעיל Pods באשכול אחר.

מידע על KubernetesPodOperator ב-Managed Airflow (דור 2)

בקטע הזה מוסבר איך KubernetesPodOperator פועל ב-Managed Airflow (דור 2).

שימוש במשאבים

ב-Managed Airflow (דור 2), האשכול של הסביבה מתרחב אוטומטית. עומסי עבודה נוספים שאתם מריצים באמצעות KubernetesPodOperator מתרחבים באופן עצמאי מהסביבה שלכם.

הסביבה שלכם לא מושפעת מהגידול בביקוש למשאבים, אבל גודל האשכול בסביבה שלכם גדל וקטן בהתאם לביקוש למשאבים.

התמחור של עומסי העבודה הנוספים שמופעלים באשכול של הסביבה מבוסס על מודל התמחור של Managed Airflow (דור 2) ועל מקו"טים של Managed Airflow Compute.

‫Managed Airflow (דור 2) משתמש באשכולות Autopilot, שכוללים את המושג compute classes:

• ב-Managed Airflow יש תמיכה רק בסוג המחשוב general-purpose.

• כברירת מחדל, אם לא נבחרה כיתה, המערכת מניחה שמדובר בכיתה general-purpose כשיוצרים תרמילים באמצעות KubernetesPodOperator.

• כל מחלקה משויכת למאפיינים ספציפיים ולמגבלות משאבים. אפשר לקרוא עליהם במסמכי Autopilot. לדוגמה, ל-Pods שפועלים בתוך המחלקה general-purpose יכולים להיות עד ‎110 GiB של זיכרון.

גישה למשאבים של הפרויקט

‫Managed Airflow (דור 2) משתמש באשכולות GKE עם איחוד זהויות של עומסי עבודה ל-GKE. לפודים שפועלים במרחב השמות composer-user-workloads יש גישה למשאבים Google Cloud בפרויקט בלי צורך בהגדרה נוספת. חשבון השירות של הסביבה שלכם משמש לגישה למשאבים האלה.

אם רוצים להשתמש במרחב שמות בהתאמה אישית, צריך למפות את חשבונות השירות ב-Kubernetes שמשויכים למרחב השמות הזה לחשבונות שירות ב- Google Cloud , כדי לאפשר אימות של זהות השירות לבקשות לממשקי API של Google ולשירותים אחרים. אם אתם מפעילים Pods במרחב שמות בהתאמה אישית באשכול של סביבת העבודה, לא נוצרים קישורי IAM בין Kubernetes לבין חשבונות שירות שלGoogle Cloud , ול-Pods האלה אין גישה למשאבים של פרויקט Google Cloud .

אם אתם משתמשים במרחב שמות בהתאמה אישית ורוצים שלקבוצות ה-Pod שלכם תהיה גישה למשאביGoogle Cloud , אתם צריכים לפעול לפי ההנחיות במאמר בנושא איחוד זהויות של עומסי עבודה ל-GKE ולהגדיר את הקישורים למרחב שמות בהתאמה אישית:

1. יוצרים מרחב שמות נפרד באשכול של הסביבה.
2. יוצרים קישור בין חשבון השירות של Kubernetes במרחב השמות המותאם אישית לבין חשבון השירות של הסביבה.
3. מוסיפים את הערת חשבון השירות של הסביבה לחשבון השירות של Kubernetes.
4. כשמשתמשים ב-KubernetesPodOperator, צריך לציין את מרחב השמות ואת חשבון השירות של Kubernetes בפרמטרים namespace ו-service_account_name.

הערה: לוקח כמה שניות עד ששרת המטא-נתונים של GKE מתחיל לקבל בקשות ב-Pod שנוצר לאחרונה. לכן, ניסיונות לאימות באמצעות איחוד זהויות של עומסי עבודה ל-GKE בתוך כמה השניות הראשונות של חיי ה-Pod עלולים להיכשל. ניסיון חוזר להתקשר יפתור את הבעיה. מידע נוסף על ההגבלה הזו זמין במאמר הגבלות של איחוד שירותי אימות הזהות של עומסי עבודה ב-GKE.

הגדרה מינימלית

כדי ליצור KubernetesPodOperator, נדרשים רק הפרמטרים name,‏ image ו-task_id של Pod. ‫/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="marketplace.gcr.io/google/ubuntu2204",
    # 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.

מידע נוסף זמין במקורות המידע הבאים:

• מידע על שימוש בסודות וב-ConfigMaps של Kubernetes זמין במאמר שימוש בסודות וב-ConfigMaps של Kubernetes.

• מידע על שימוש בתבניות Jinja עם KubernetesPodOperator זמין במאמר שימוש בתבניות Jinja.

• מידע על הפרמטרים של 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 תומך בתבניות Jinja ב-DAG.

צריך להצהיר על פרמטרים נדרשים של Airflow ‏ (task_id,‏ name ו-image) באמצעות האופרטור. כמו שרואים בדוגמה הבאה, אפשר להשתמש ב-Jinja כדי ליצור תבנית לכל הפרמטרים האחרים, כולל cmds,‏ arguments,‏ env_vars ו-config_file.

הפרמטר env_vars בדוגמה מוגדר ממשתנה Airflow בשם my_value. ערך ה-DAG בדוגמה מתקבל ממשתנה התבנית vars ב-Airflow. ל-Airflow יש עוד משתנים שמאפשרים גישה לסוגים שונים של מידע. לדוגמה, אפשר להשתמש במשתנה התבנית conf כדי לגשת לערכים של אפשרויות ההגדרה של Airflow. מידע נוסף ורשימת המשתנים שזמינים ב-Airflow מופיעים בחומר העזר בנושא תבניות במסמכי התיעוד של Airflow.

בלי לשנות את ה-DAG או ליצור את המשתנה env_vars, המשימה ex-kube-templates בדוגמה נכשלת כי המשתנה לא קיים. יוצרים את המשתנה הזה בממשק המשתמש של Airflow או באמצעות Google Cloud CLI:

ממשק המשתמש של Airflow

1. עוברים אל ממשק המשתמש של Airflow.

2. בסרגל הכלים, בוחרים באפשרות אדמין > משתנים.

3. בדף List Variable (משתנה רשימה), לוחצים על Add a new record (הוספת רשומה חדשה).

4. בדף Add Variable (הוספת משתנה), מזינים את הפרטים הבאים:

   • מקש:my_value
   • ערך: example_value

5. לוחצים על Save.

gcloud

מזינים את הפקודה הבאה:

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

מחליפים את:

• ‫ENVIRONMENT בשם הסביבה.
• ‫LOCATION עם האזור שבו הסביבה ממוקמת.

בדוגמה הבאה אפשר לראות איך משתמשים בתבניות Jinja עם 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",
)

שימוש בסודות וב-ConfigMaps של Kubernetes

סוד ב-Kubernetes הוא אובייקט שמכיל מידע רגיש. ‫ConfigMap ב-Kubernetes הוא אובייקט שמכיל נתונים לא סודיים בצמדי מפתח-ערך.

ב-Managed Airflow (דור 2), אפשר ליצור Secrets ו-ConfigMaps באמצעות Google Cloud CLI,‏ API או Terraform, ואז לגשת אליהם מ-KubernetesPodOperator.

מידע על קובצי הגדרות ב-YAML

כשיוצרים Kubernetes Secret או ConfigMap באמצעות Google Cloud CLI ו-API, צריך לספק קובץ בפורמט YAML. הפורמט של הקובץ הזה צריך להיות זהה לפורמט שבו משתמשים ב-Kubernetes Secrets וב-ConfigMaps. בתיעוד של Kubernetes יש הרבה דוגמאות קוד של ConfigMap ו-Secrets. כדי להתחיל, אפשר לעיין בדף הפצת פרטי כניסה בצורה מאובטחת באמצעות סודות ובמאמר בנושא ConfigMaps.

כמו ב-Kubernetes Secrets, צריך להשתמש בייצוג base64 כשמגדירים ערכים ב-Secrets.

זהירות: ערכים שמקודדים ב-Base64 לא מוצפנים בשום צורה, אלא רק מוסתרים. אפשר לשחזר את הערך המקורי מהייצוג שלו ב-base64.

כדי לקודד ערך, אפשר להשתמש בפקודה הבאה (זו אחת מתוך הרבה דרכים לקבל ערך מקודד בפורמט Base64):

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

פלט:

cG9zdGdyZXNxbCtwc3ljb3BnMjovL3Jvb3Q6ZXhhbXBsZS1wYXNzd29yZEAxMjcuMC4wLjE6MzMwNi9leGFtcGxlLWRiIC1uCg==

בהמשך המדריך הזה נשתמש בשתי הדוגמאות הבאות של קובצי YAML. קובץ הגדרות לדוגמה ב-YAML בשביל סוד של Kubernetes:

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

קובץ הגדרות לדוגמה ב-YAML עבור ConfigMap. אין צורך להשתמש בייצוג base64 ב-ConfigMaps:

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

ניהול סודות ב-Kubernetes

ב-Managed Airflow (דור 2), יוצרים סודות באמצעות Google Cloud CLI ו-kubectl:

1. קבלת מידע על האשכול בסביבה:

   1. מריצים את הפקודה הבאה:

      gcloud composer environments describe ENVIRONMENT \
          --location LOCATION \
          --format="value(config.gkeCluster)"
      

      מחליפים את:

      • ENVIRONMENT בשם של הסביבה.
      • מחליפים את LOCATION באזור שבו נמצאת סביבת Managed Airflow.

      הפלט של הפקודה הזו הוא בפורמט הבא: projects/<your-project-id>/locations/<location-of-composer-env>/clusters/<your-cluster-id>.

   2. כדי לקבל את מזהה אשכול GKE, מעתיקים את הפלט אחרי /clusters/ (מסתיים ב--gke).

2. מתחברים לאשכול GKE באמצעות הפקודה הבאה:

   gcloud container clusters get-credentials CLUSTER_ID \
     --project PROJECT \
     --region LOCATION
   

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫CLUSTER_ID: מזהה האשכול של הסביבה.
   • ‫PROJECT_ID: מזהה הפרויקט.
   • ‫LOCATION: האזור שבו נמצאת הסביבה.

3. יצירת סודות של Kubernetes:

   הערה: במאמר ניהול סודות באמצעות kubectl מפורטות פקודות לעריכה ולמחיקה של סודות.

   הפקודות הבאות מדגימות שתי גישות שונות ליצירת סודות של Kubernetes. בגישה --from-literal משתמשים בצמדי מפתח/ערך. הגישה --from-file משתמשת בתוכן הקובץ.

   • כדי ליצור Kubernetes Secret על ידי ציון צמדי מפתח/ערך, מריצים את הפקודה הבאה. בדוגמה הזו נוצר סוד בשם airflow-secrets עם שדה sql_alchemy_conn והערך test_value.

     kubectl create secret generic airflow-secrets \
       --from-literal sql_alchemy_conn=test_value -n composer-user-workloads
     

   • כדי ליצור סוד של Kubernetes על ידי ציון תוכן הקובץ, מריצים את הפקודה הבאה. בדוגמה הזו נוצר סוד בשם service-account עם השדה service-account.json שהערך שלו נלקח מהתוכן של קובץ מקומי בשם ./key.json.

     kubectl create secret generic service-account \
       --from-file service-account.json=./key.json -n composer-user-workloads
     

שימוש ב-Kubernetes Secrets ב-DAG

בדוגמה הזו מוצגות שתי דרכים להשתמש ב-Kubernetes Secrets: כמשתנה סביבתי וכנפח שמוטמע על ידי ה-Pod.

הסוד הראשון, airflow-secrets, מוגדר כמשתנה סביבה של Kubernetes בשם SQL_CONN (בניגוד למשתנה סביבה של Airflow או Managed Airflow).

הסוד השני, 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_env. הסוד הזה נקרא airflow-secrets. הפרמטר deploy_type מציין שהמשתנה צריך להיות חשוף כמשתנה סביבה. השם של משתנה הסביבה הוא SQL_CONN, כפי שצוין בפרמטר deploy_target. לבסוף, הערך של משתנה הסביבה SQL_CONN מוגדר לערך של המפתח sql_alchemy_conn.

השם של הסוד השני ב-Kubernetes מוגדר במשתנה secret_volume. הסוד הזה נקרא service-account. הוא נחשף כנפח, כפי שמצוין בפרמטר deploy_type. הנתיב של הקובץ שצריך לטעון, deploy_target, הוא /var/secrets/google. לבסוף, הערך key של הסוד שמאוחסן ב-deploy_target הוא service-account.json.

כך נראית הגדרת האופרטור:

kubernetes_secret_vars_ex = KubernetesPodOperator(
    task_id="ex-kube-secrets",
    name="ex-kube-secrets",
    namespace="composer-user-workloads",
    image="marketplace.gcr.io/google/ubuntu2204",
    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",
)

מידע על ספק CNCF Kubernetes

האופרטור KubernetesPodOperator מיושם בספק apache-airflow-providers-cncf-kubernetes.

הערות מפורטות על הגרסה של ספק CNCF Kubernetes זמינות באתר של ספק CNCF Kubernetes.

גרסה 6.0.0

בגרסה 6.0.0 של חבילת הספק CNCF Kubernetes, נעשה שימוש בחיבור kubernetes_default כברירת מחדל ב-KubernetesPodOperator.

אם ציינתם חיבור בהתאמה אישית בגרסה 5.0.0, המפעיל עדיין משתמש בחיבור הזה. כדי לחזור לשימוש בחיבור kubernetes_default, יכול להיות שתצטרכו לשנות את הגדרות ה-DAG בהתאם.

גרסה 5.0.0

בגרסה הזו יש כמה שינויים שלא תואמים לאחור בהשוואה לגרסה 4.4.0. החשובים ביותר קשורים לkubernetes_defaultחיבור שלא נמצא בשימוש בגרסה 5.0.0.

חשוב: אם אתם משדרגים מגרסה של Managed Airflow שמשתמשת בגרסה 4.4.0 או בגרסה קודמת של ספק לגרסה של Managed Airflow שמשתמשת בגרסה 5.0.0, תצטרכו לבצע שינויים ב-DAG באמצעות KubernetesPodOperator.

• צריך לשנות את החיבור של kubernetes_default. נתיב ההגדרה של Kubernetes צריך להיות מוגדר ל-/home/airflow/composer_kube_config (כמו שמוצג באיור הבא). לחלופין, צריך להוסיף את config_file להגדרות של KubernetesPodOperator (כמו בדוגמת הקוד הבאה).

איור 1. ממשק המשתמש של Airflow, שינוי החיבור kubernetes_default (לחיצה להגדלה)

• כדי לשנות את הקוד של משימה באמצעות KubernetesPodOperator, פועלים לפי השלבים הבאים:

KubernetesPodOperator(
  # config_file parameter - can be skipped if connection contains this setting
  config_file="/home/airflow/composer_kube_config",
  # definition of connection to be used by the operator
  kubernetes_conn_id='kubernetes_default',
  ...
)

מידע נוסף על גרסה 5.0.0 זמין בהערות המוצר של ספק Kubernetes של CNCF.

פתרון בעיות

בקטע הזה מפורטות הצעות לפתרון בעיות נפוצות שקשורות ל-KubernetesPodOperator:

צפייה ביומנים

כשמנסים לפתור בעיות, אפשר לבדוק את היומנים בסדר הבא:

1. יומני משימות של Airflow:

   1. נכנסים לדף Environments במסוף Google Cloud .

      מעבר אל Environments

   2. ברשימת הסביבות, לוחצים על שם הסביבה. הדף Environment details ייפתח.

   3. עוברים לכרטיסייה DAGs.

   4. לוחצים על השם של ה-DAG ואז על ההרצה של ה-DAG כדי לראות את הפרטים והיומנים.

2. יומנים של מתזמן Airflow:

   1. עוברים לדף פרטי הסביבה.

   2. עוברים לכרטיסייה יומנים.

   3. בודקים את היומנים של מתזמן Airflow.

3. יומני Pod במסוף Google Cloud , בקטע GKE workloads. היומנים האלה כוללים את קובץ ה-YAML של הגדרת ה-Pod, אירועי ה-Pod ופרטי ה-Pod.

קודי החזרה שאינם אפס

כשמשתמשים ב-KubernetesPodOperator (וב-GKEStartPodOperator), קוד החזרה של נקודת הכניסה של הקונטיינר קובע אם המשימה נחשבת למוצלחת או לא. קודי החזרה שאינם אפס מציינים כשל.

דפוס נפוץ הוא להריץ סקריפט של מעטפת כנקודת הכניסה של הקונטיינר כדי לקבץ כמה פעולות בתוך הקונטיינר.

אם אתם כותבים סקריפט כזה, מומלץ לכלול את הפקודה set -e בראש הסקריפט, כדי שפקודות שנכשלו בסקריפט יסיימו את הסקריפט ויעבירו את הכשל למופע המשימה של Airflow.

הזמן הקצוב לתפוגה של Pod

זמן הקצוב לתפוגה (timeout) שמוגדר כברירת מחדל ב-KubernetesPodOperator הוא 120 שניות, מה שיכול לגרום לזמני קצוב לתפוגה לפני הורדה של תמונות גדולות יותר. אפשר להגדיל את הזמן הקצוב לתפוגה באמצעות שינוי הפרמטר startup_timeout_seconds כשיוצרים את KubernetesPodOperator.

כשפוד מגיע לזמן קצוב לתפוגה, היומן הספציפי למשימה זמין בממשק המשתמש של Airflow. לדוגמה:

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

יכול להיות שפסק זמן של Pod יתרחש גם אם ל-Managed Airflow חסרות הרשאות ה-IAM הנדרשות לביצוע המשימה. כדי לוודא זאת, אפשר להשתמש בלוחות הבקרה של GKE כדי לבדוק את היומנים של עומס העבודה הספציפי, או להשתמש ב-Cloud Logging.

יצירת חיבור חדש נכשלה

השדרוג האוטומטי מופעל כברירת מחדל באשכולות GKE. אם מאגר הצמתים נמצא באשכול שעובר שדרוג, יכול להיות שתופיע השגיאה הבאה:

<Task(KubernetesPodOperator): gke-upgrade> Failed to establish a new
connection: [Errno 111] Connection refused

כדי לבדוק אם האשכול עובר שדרוג, במסוף Google Cloud , עוברים לדף Kubernetes clusters ומחפשים את סמל הטעינה לצד שם האשכול של הסביבה.

המאמרים הבאים

• שימוש באופרטורים של GKE
• מידע בסיסי על Kubernetes