כיצד להפעיל DAG ב-Airflow באמצעות ה-API הניסיוני

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

זה היה המקרה, למשל, בשנת 2015, ובמהלך התוכנית "Big Data Specialist" השתמשנו באשכול Hadoop עם Spark עבור 35 משתמשים בו זמנית. לא היה ברור כיצד להכין אותו למקרה שימוש כזה באמצעות YARN. בסופו של דבר, לאחר שהבנו את זה והלכנו בשביל בעצמנו, עשינו זאת פוסט על Habré וגם הופיע ב מוסקבה ספארק מפגש.

פרהיסטוריה

הפעם נדבר על תוכנית אחרת - מהנדס נתונים. המשתתפים שלנו בונים עליו שני סוגי ארכיטקטורה: למבדה וקאפה. ובארכיטקטורת lamdba, כחלק מעיבוד אצווה, נעשה שימוש ב-Airflow להעברת יומנים מ-HDFS ל-ClickHouse.

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

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

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

לאחר מכן, נתאר את כל הנתיב: מהתקנת Airflow ועד ליצירת בקשת POST שמפעילה את ה-DAG באמצעות ה-API הניסיוני. אנו נעבוד עם אובונטו 16.04.

1. התקנת זרימת אוויר

בוא נבדוק שיש לנו Python 3 ו-virtualenv.

$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0

אם משהו מזה חסר, אז התקן אותו.

כעת ניצור ספרייה בה נמשיך לעבוד עם Airflow.

$ mkdir <your name of directory>
$ cd /path/to/your/new/directory
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $

התקן זרימת אוויר:

(venv) $ pip install airflow

הגרסה עליה עבדנו: 1.10.

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

(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=<path to airflow_home>

השלב הבא הוא להפעיל פקודה שתיצור ותאתחל מסד נתונים של זרימת נתונים ב- SQLite:

(venv) $ airflow initdb

מסד הנתונים ייווצר ב airflow.db כברירת מחדל.

בוא נבדוק אם Airflow מותקן:

$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Using executor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/PatternGrammar.txt
  ____________       _____________
 ____    |__( )_________  __/__  /________      __
____  /| |_  /__  ___/_  /_ __  /_  __ _ | /| / /
___  ___ |  / _  /   _  __/ _  / / /_/ /_ |/ |/ /
 _/_/  |_/_/  /_/    /_/    /_/  ____/____/|__/
   v1.10.0

אם הפקודה עבדה, אז Airflow יצרה קובץ תצורה משלה airflow.cfg в AIRFLOW_HOME:

$ tree
.
├── airflow.cfg
└── unittests.cfg

לזרימת האוויר יש ממשק אינטרנט. ניתן להפעיל אותו על ידי הפעלת הפקודה:

(venv) $ airflow webserver --port 8081

כעת תוכל להקיש על ממשק האינטרנט בדפדפן ביציאה 8081 במארח שבו פעל Airflow, לדוגמה: <hostname:8081>.

2. עבודה עם API ניסיוני

בשלב זה, זרימת האוויר מוגדרת ומוכנה לפעולה. עם זאת, עלינו להפעיל גם את ה-API הניסיוני. הדמקה שלנו כתובה ב-Python, כך שבהמשך כל הבקשות יהיו בו באמצעות הספרייה requests.

למעשה, ה-API כבר עובד עבור בקשות פשוטות. לדוגמה, בקשה זו מאפשרת לך לבדוק את פעולתה:

>>> import requests
>>> host = <your hostname>
>>> airflow_port = 8081 #в нашем случае такой, а по дефолту 8080
>>> requests.get('http://{}:{}/{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'

אם אתה מקבל הודעה כזו בתגובה, זה אומר שהכל עובד.

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

לשם כך, תצטרך לבצע מספר שלבים נוספים.

ראשית, עליך להוסיף את זה לתצורה:

[api]
auth_backend = airflow.contrib.auth.backends.password_auth

לאחר מכן, עליך ליצור את המשתמש שלך עם זכויות אדמין:

>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.Admin())
>>> user.username = 'new_user_name'
>>> user.password = 'set_the_password'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()

לאחר מכן, עליך ליצור משתמש עם זכויות רגילות שיורשה להפעיל את ה-DAG.

>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.User())
>>> user.username = 'newprolab'
>>> user.password = 'Newprolab2019!'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()

עכשיו הכל מוכן.

3. הפעל בקשת POST

בקשת ה-POST עצמה תיראה כך:

>>> dag_id = newprolab
>>> url = 'http://{}:{}/{}/{}/{}'.format(host, airflow_port, 'api/experimental/dags', dag_id, 'dag_runs')
>>> data = {"conf":"{"key":"value"}"}
>>> headers = {'Content-type': 'application/json'}
>>> auth = ('newprolab', 'Newprolab2019!')
>>> uri = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
>>> uri.text
'{n  "message": "Created <DagRun newprolab @ 2019-03-27 10:24:25+00:00: manual__2019-03-27T10:24:25+00:00, externally triggered: True>"n}n'

הבקשה טופלה בהצלחה.

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

הבדיקה הושלמה.

מקור: www.habr.com