Airflow הוא כלי לפיתוח ותחזוקה נוחה ומהירה של תהליכי עיבוד נתונים אצווה

שלום, הבר! במאמר זה אני רוצה לדבר על כלי אחד נהדר לפיתוח תהליכי עיבוד נתונים אצווה, למשל, בתשתית של DWH ארגוני או DataLake שלך. נדבר על Apache Airflow (להלן Airflow). היא נשללת באופן לא הוגן תשומת לב על Habré, ובעיקר אנסה לשכנע אותך שלפחות Airflow שווה להסתכל כאשר אתה בוחר מתזמן לתהליכי ה-ETL/ELT שלך.

בעבר כתבתי סדרת מאמרים בנושא DWH כשעבדתי בבנק טינקוף. כעת הפכתי לחלק מצוות Mail.Ru Group ואני מפתח פלטפורמה לניתוח נתונים בתחום המשחקים. למעשה, כשיופיעו חדשות ופתרונות מעניינים, הצוות שלי ואני נדבר כאן על הפלטפורמה שלנו לניתוח נתונים.

פרולוג

אז בואו נתחיל. מהי זרימת אוויר? זוהי ספרייה (או סט ספריות) לפתח, לתכנן ולנטר תהליכי עבודה. התכונה העיקרית של Airflow: קוד Python משמש לתיאור (לפתח) תהליכים. יש לזה הרבה יתרונות לארגון הפרויקט והפיתוח שלך: בעצם, פרויקט ה-ETL שלך (למשל) הוא רק פרויקט Python, ואתה יכול לארגן אותו כרצונך, תוך התחשבות בפרטי התשתית, גודל הצוות ו דרישות אחרות. מבחינה אינסטרומנטלית הכל פשוט. השתמש למשל PyCharm + Git. זה נפלא ומאוד נוח!

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

DAG

DAG הוא איזשהו שיוך משמעותי של המשימות שלך שאתה רוצה להשלים ברצף מוגדר בהחלט לפי לוח זמנים ספציפי. Airflow מספק ממשק אינטרנט נוח לעבודה עם DAGs וישויות אחרות:

ה-DAG עשוי להיראות כך:

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

אופרטורים

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

• BashOperator - אופרטור לביצוע פקודת bash.
• PythonOperator - מפעיל לקריאה לקוד Python.
• EmailOperator — אופרטור לשליחת דואר אלקטרוני.
• HTTPOperator - אופרטור לעבודה עם בקשות http.
• SqlOperator - אופרטור לביצוע קוד SQL.
• סנסור הוא מפעיל להמתנה לאירוע (הגעת השעה הנדרשת, הופעת הקובץ הנדרש, שורה במאגר, תגובה מה-API וכו' וכו').

ישנם אופרטורים ספציפיים יותר: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

אתה יכול גם לפתח אופרטורים על סמך המאפיינים שלך ולהשתמש בהם בפרויקט שלך. לדוגמה, יצרנו את MongoDBToHiveViaHdfsTransfer, אופרטור לייצוא מסמכים מ-MongoDB ל-Hive, וכמה אופרטורים לעבודה עם קליקהאוס: CHLoadFromHiveOperator ו-CHTableLoaderOperator. בעיקרו של דבר, ברגע שפרויקט השתמש לעתים קרובות בקוד הבנוי על הצהרות בסיסיות, אתה יכול לחשוב על בנייתו להצהרה חדשה. זה יפשט את המשך הפיתוח, ותרחיב את ספריית המפעילים שלך בפרויקט.

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

מתזמן

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

לכל בריכה יש הגבלה על מספר המשבצות. בעת יצירת DAG, הוא מקבל מאגר:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

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

1. המשימות הקודמות הושלמו ב-DAG; משימות חדשות ניתן לעמוד בתור.
2. התור ממוין בהתאם לעדיפות המשימות (ניתן גם לשלוט בסדרי העדיפויות), ואם יש משבצת פנויה בבריכה, ניתן לקחת את המשימה לפעולה.
3. אם יש סלרי עובד חינם, המשימה נשלחת אליו; העבודה שתכנתת בבעיה מתחילה, באמצעות אופרטור כזה או אחר.

פשוט מספיק.

מתזמן פועל על הסט של כל DAGs וכל המשימות בתוך DAGs.

כדי שמתזמן יתחיל לעבוד עם DAG, ה-DAG צריך להגדיר לוח זמנים:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

יש קבוצה של הגדרות מוכנות מראש: @once, @hourly, @daily, @weekly, @monthly, @yearly.

אתה יכול גם להשתמש בביטויי cron:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

תאריך ביצוע

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

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

יצירת משימות

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

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

ה-DAG נראה כך:

במקרה זה, אתה יכול להוסיף או להסיר רסיס פשוט על ידי התאמת ההגדרות ועדכון ה-DAG. נוֹחַ!

אתה יכול גם להשתמש ביצירת קוד מורכב יותר, למשל, לעבוד עם מקורות בצורה של מסד נתונים או לתאר מבנה טבלה, אלגוריתם לעבודה עם טבלה, ובהתחשב בתכונות של תשתית DWH, ליצור תהליך לטעינת N טבלאות לאחסון שלך. או, למשל, עבודה עם API שאינו תומך בעבודה עם פרמטר בצורת רשימה, ניתן ליצור N משימות ב-DAG מתוך רשימה זו, להגביל את ההקבלה של בקשות ב-API למאגר ולגרד הנתונים הדרושים מה-API. גָמִישׁ!

מאגר

ל-Airflow יש מאגר אחורי משלה, מסד נתונים (יכול להיות MySQL או Postgres, יש לנו Postgres), המאחסן את מצבי המשימות, DAGs, הגדרות חיבור, משתנים גלובליים וכו' וכו'. כאן הייתי רוצה לומר ש- מאגר ב-Airflow הוא פשוט מאוד (בערך 20 טבלאות) ונוח אם אתה רוצה לבנות תהליכים משלך על גביו. אני זוכר את 100500 הטבלאות במאגר אינפורמטיקה, שהיה צריך ללמוד הרבה זמן לפני שהבינו איך לבנות שאילתה.

ניטור

לאור הפשטות של המאגר, אתה יכול לבנות תהליך ניטור משימות שנוח לך. אנו משתמשים בפנקס רשימות ב-Zeppelin, שבו אנו בוחנים את מצב המשימות:

זה יכול להיות גם ממשק האינטרנט של Airflow עצמו:

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

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

בסך הכל

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

עכשיו יש לנו Airflow עובד מדי יום כ-6,5 אלף משימות. הם די שונים באופיים. ישנן משימות של טעינת נתונים ל-DWH הראשי ממקורות רבים ושונים ומאוד ספציפיים, יש משימות של חישוב חלונות ראווה בתוך ה-DWH הראשי, יש משימות של פרסום נתונים ל-DWH מהיר, יש הרבה מאוד משימות שונות - ו-Airflow לועס את כולם יום אחר יום. אם מדברים במספרים, זהו 2,3 אלף משימות ELT במורכבות משתנה בתוך DWH (Hadoop), בערך. 2,5 מאות מאגרי מידע מקורות, זה צוות מ 4 מפתחי ETL, אשר מחולקים לעיבוד נתונים ETL ב-DWH ועיבוד נתונים ELT בתוך DWH וכמובן עוד מנהל אחד, העוסק בתשתית השירות.

תוכניות לעתיד

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

אפילוג

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

מקור: www.habr.com