प्रायोगिक API वापरून एअरफ्लोमध्ये DAG ट्रिगर कसा बनवायचा

आमचे शैक्षणिक कार्यक्रम तयार करताना, आम्हाला ठराविक साधनांसह काम करताना वेळोवेळी अडचणी येतात. आणि या क्षणी जेव्हा आम्ही त्यांचा सामना करतो तेव्हा नेहमीच पुरेसे दस्तऐवज आणि लेख नसतात जे आम्हाला या समस्येचा सामना करण्यास मदत करतील.

उदाहरणार्थ, 2015 मध्ये ही परिस्थिती होती आणि “बिग डेटा स्पेशलिस्ट” प्रोग्राम दरम्यान आम्ही एकाच वेळी 35 वापरकर्त्यांसाठी स्पार्कसह हॅडूप क्लस्टर वापरला. यार्न वापरून अशा वापराच्या केससाठी ते कसे तयार करायचे ते स्पष्ट नव्हते. शेवटी, ते शोधून काढले आणि स्वतःच्या मार्गावर चालत गेलो Habré वर पोस्ट आणि येथे देखील सादर केले मॉस्को स्पार्क मीटअप.

prehistory

यावेळी आपण एका वेगळ्या कार्यक्रमाबद्दल बोलू - डेटा इंजिनियर. आमचे सहभागी त्यावर दोन प्रकारचे आर्किटेक्चर तयार करतात: लॅम्बडा आणि कप्पा. आणि lamdba आर्किटेक्चरमध्ये, बॅच प्रक्रियेचा एक भाग म्हणून, HDFS वरून क्लिकहाऊसमध्ये लॉग हस्तांतरित करण्यासाठी एअरफ्लोचा वापर केला जातो.

सर्व काही साधारणपणे चांगले आहे. त्यांना स्वतःच्या पाइपलाइन बांधू द्या. तथापि, तेथे एक "पण" आहे: आमचे सर्व कार्यक्रम हे शिकण्याच्या प्रक्रियेच्या दृष्टिकोनातून तांत्रिकदृष्ट्या प्रगत आहेत. लॅब तपासण्यासाठी, आम्ही स्वयंचलित चेकर्स वापरतो: सहभागीला त्याच्या वैयक्तिक खात्यावर जाणे आवश्यक आहे, "चेक" बटणावर क्लिक करा आणि काही काळानंतर त्याला त्याने काय केले यावर काही प्रकारचे विस्तारित अभिप्राय दिसतो. आणि या क्षणीच आपण आपल्या समस्येकडे जाण्यास सुरवात करतो.

या लॅबची पडताळणी अशी रचना केली आहे: आम्ही सहभागीच्या काफ्काला एक नियंत्रण डेटा पॅकेट पाठवतो, त्यानंतर Gobblin हे डेटा पॅकेट HDFS मध्ये हस्तांतरित करतो, त्यानंतर Airflow हे डेटा पॅकेट घेतो आणि क्लिकहाऊसमध्ये ठेवतो. युक्ती अशी आहे की एअरफ्लोला हे रिअल टाइममध्ये करण्याची आवश्यकता नाही, ते एका शेड्यूलनुसार करते: प्रत्येक 15 मिनिटांनी फायलींचा एक समूह घेते आणि त्या अपलोड करते.

हे निष्पन्न झाले की येथे आणि आता तपासक चालू असताना आम्हाला आमच्या विनंतीनुसार त्यांचे DAG स्वतःच ट्रिगर करावे लागेल. गुगलिंग केल्यानंतर, आम्हाला आढळले की एअरफ्लोच्या नंतरच्या आवृत्त्यांसाठी एक तथाकथित आहे प्रायोगिक API. शब्द experimental, नक्कीच, भीतीदायक वाटते, पण काय करावे... अचानक ते बंद होते.

पुढे, आम्ही संपूर्ण मार्गाचे वर्णन करू: एअरफ्लो स्थापित करण्यापासून ते प्रायोगिक API वापरून DAG ला ट्रिगर करणारी POST विनंती तयार करण्यापर्यंत. आम्ही उबंटू 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_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 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.cfg в AIRFLOW_HOME:

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

एअरफ्लोमध्ये वेब इंटरफेस आहे. हे कमांड चालवून लॉन्च केले जाऊ शकते:

(venv) $ airflow webserver --port 8081

तुम्ही आता होस्टवर पोर्ट 8081 वर ब्राउझरमध्ये वेब इंटरफेस दाबू शकता जेथे Airflow चालू होते, उदाहरणार्थ: <hostname:8081>.

2. प्रायोगिक API सह कार्य करणे

या टप्प्यावर, एअरफ्लो कॉन्फिगर केले आहे आणि जाण्यासाठी तयार आहे. तथापि, आम्हाला प्रायोगिक API देखील चालवावे लागेल. आमचे चेकर्स पायथनमध्ये लिहिलेले आहेत, त्यामुळे पुढील सर्व विनंत्या लायब्ररी वापरून त्यामध्ये असतील 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 ला प्रक्रिया करण्यासाठी थोडा वेळ देतो आणि क्लिकहाऊस टेबलला विनंती करतो, नियंत्रण डेटा पॅकेट पकडण्याचा प्रयत्न करतो.

चेक पूर्ण झाला.

स्त्रोत: www.habr.com