פתרון בעיות בעדכונים ובשדרוגים של סביבות

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

בדף הזה מפורט מידע לפתרון בעיות שעלולות לקרות במהלך עדכון או שדרוג של סביבות Managed Service for Apache Airflow.

מידע על פתרון בעיות שקשורות ליצירת סביבות מופיע במאמר פתרון בעיות ביצירת סביבות.

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

• בעיות בהרשאות של חשבון שירות
• בעיות בתלות ב-PyPI
• גודל מסד הנתונים של Airflow

אין הרשאות מספיקות לעדכן או לשדרג סביבה

אם ל-Managed Airflow אין מספיק הרשאות כדי לעדכן או לשדרג סביבה, מוצגת הודעת השגיאה הבאה:

ERROR: (gcloud.composer.environments.update) PERMISSION_DENIED: The caller does not have permission

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

לחשבון השירות של הסביבה אין הרשאות מספיקות

כשיוצרים סביבת Managed Airflow, מציינים חשבון שירות שמבצע את רוב הפעולות בסביבה. אם לחשבון השירות הזה אין מספיק הרשאות לפעולה המבוקשת, המערכת של Managed Airflow מציגה שגיאה:

    UPDATE operation on this environment failed 3 minutes ago with the
    following error message:
    Composer Backend timed out. Currently running tasks are [stage:
    CP_COMPOSER_AGENT_RUNNING
    description: "No agent response published."
    response_timestamp {
      seconds: 1618203503
      nanos: 291000000
    }
    ].

הפתרון: מקצים תפקידים לחשבון Google ולחשבון השירות של הסביבה, כמו שמתואר במאמר בנושא בקרת גישה.

מסד הנתונים של Airflow גדול מדי ואי אפשר לבצע את הפעולה

יכול להיות שפעולת שדרוג לא תצליח כי גודל מסד הנתונים של Airflow גדול מדי.

אם גודל מסד הנתונים של Airflow גדול מ-20GB,‏ Managed Airflow מציג את השגיאה הבאה:

Airflow database uses more than 20 GB. Please clean the database before upgrading.

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

שדרוג לגרסה חדשה של Managed Airflow נכשל בגלל קונפליקטים בחבילת PyPI

כשמשדרגים סביבה עם חבילות PyPI מותאמות אישית שהותקנו, יכול להיות שתיתקלו בשגיאות שקשורות לקונפליקטים בחבילות PyPI. יכול להיות שהסיבה לכך היא שב-Airflow build החדש יש גרסאות עדכניות יותר של חבילות שהותקנו מראש. זה יכול לגרום לקונפליקטים של תלות בחבילות PyPI שהתקנתם בסביבה שלכם.

הפתרון:

• כדי לקבל מידע מפורט על התנגשויות בין חבילות, מריצים בדיקת שדרוג.
• הגבלות הגרסה של חבילות PyPI מותאמות אישית שהותקנו. לדוגמה, במקום לציין גרסה בתור ==1.0.1, מציינים אותה בתור >=1.0.1.
• לקבלת מידע נוסף על שינוי דרישות הגרסה כדי לפתור בעיות של תלות סותרת, אפשר לעיין במסמכי העזרה של pip.

בדיקת אזהרות על העברה שנכשלה

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

הפתרון:

אפשר להשתמש בשני ה-DAG הבאים כדי לבדוק את הנתונים שהועברו ולשנות את השם של הטבלאות.

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

אם אתם לא צריכים את הנתונים או שכבר תיקנתם אותם, אתם יכולים להריץ את rename_moved_tables_after_upgrade_dag DAG. ה-DAG הזה משנה את השם של הטבלאות שהועברו. הטבלאות והנתונים שלהן לא נמחקים, כך שתוכלו לבדוק את הנתונים בשלב מאוחר יותר.

"""
When upgrading Airflow to a newer version,
it might happen that some data cannot be migrated,
often because of constraint changes in the metadata base.
This file contains 2 DAGs:

1. 'list_moved_tables_after_upgrade_dag'
  Prints the rows which failed to be migrated.
2. 'rename_moved_tables_after_upgrade_dag'
  Renames the table which contains the failed migrations. This will remove the
  warning message from airflow.
"""

import datetime
import logging

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.settings import AIRFLOW_MOVED_TABLE_PREFIX


def get_moved_tables():
    hook = PostgresHook(postgres_conn_id="airflow_db")
    return hook.get_records(
        "SELECT schemaname, tablename FROM pg_catalog.pg_tables WHERE tablename"
        f" LIKE '{AIRFLOW_MOVED_TABLE_PREFIX}_%'"
    )


def list_moved_records():
    tables = get_moved_tables()
    if not tables:
        logging.info("No moved tables found")
        return

    hook = PostgresHook(postgres_conn_id="airflow_db")
    for schema, table in tables:
        df = hook.get_pandas_df(f"SELECT * FROM {schema}.{table}")
        logging.info(df.to_markdown())


def rename_moved_tables():
    tables = get_moved_tables()
    if not tables:
        return

    hook = PostgresHook(postgres_conn_id="airflow_db")
    for schema, table in tables:
        hook.run(f"ALTER TABLE {schema}.{table} RENAME TO _abandoned_{table}")


with DAG(
    dag_id="list_moved_tables_after_upgrade_dag",
    start_date=datetime.datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
):
    t1 = PythonOperator(
        task_id="list_moved_records", python_callable=list_moved_records
    )

with DAG(
    dag_id="rename_moved_tables_after_upgrade_dag",
    start_date=datetime.datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:
    t1 = PythonOperator(
        task_id="rename_moved_tables", python_callable=rename_moved_tables
    )

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

• עדכון סביבות
• שדרוג סביבות
• פתרון בעיות ביצירת סביבה