प्रायोगिक एपीआई का उपयोग करके एयरफ्लो में डीएजी ट्रिगर कैसे करें

अपने शैक्षिक कार्यक्रमों को तैयार करते समय, हम कुछ उपकरणों के साथ काम करने के मामले में समय-समय पर कठिनाइयों का सामना करते हैं। और उस क्षण जब हम उनका सामना करते हैं, तो हमेशा पर्याप्त दस्तावेज और लेख नहीं होते हैं जो इस समस्या से निपटने में मदद करेंगे।

यह मामला था, उदाहरण के लिए, 2015 में, और हम, बिग डेटा स्पेशलिस्ट प्रोग्राम पर, 35 एक साथ उपयोगकर्ताओं के लिए स्पार्क के साथ एक हडोप क्लस्टर का इस्तेमाल किया। YARN का उपयोग करके ऐसे उपयोगकर्ता मामले के तहत इसे कैसे पकाने के लिए स्पष्ट नहीं था। नतीजतन, पता लगा और अपने दम पर रास्ता चल दिया, उन्होंने हैबे पर एक पोस्ट किया और मॉस्को स्पार्क मीटअप में भी प्रदर्शन किया।

प्रागितिहास

इस बार हम एक और कार्यक्रम - डाटा इंजीनियर के बारे में बात करेंगे। हमारे प्रतिभागी इस पर दो प्रकार की वास्तुकला का निर्माण करते हैं: लंबोदा और कप्पा। और लैम्ब्डा आर्किटेक्चर में, बैच प्रोसेसिंग के हिस्से के रूप में, एचडीएफएस से क्लिकहाउस में लॉग्स ट्रांसफर करने के लिए एयरफ्लो का उपयोग किया जाता है।

सब कुछ सामान्य रूप से अच्छा है। उन्हें अपनी पाइपलाइन बनाने दें। हालाँकि, एक "लेकिन" है: हमारे सभी कार्यक्रम सीखने की प्रक्रिया के संदर्भ में तकनीकी हैं। लैब की जांच करने के लिए, हम स्वचालित चेकर्स का उपयोग करते हैं: प्रतिभागी को अपने व्यक्तिगत खाते में जाने की जरूरत है, "चेक" बटन पर क्लिक करें, और थोड़ी देर के बाद उसने जो किया है उस पर किसी प्रकार की विस्तारित प्रतिक्रिया देखता है। और यह इस समय है कि हम अपनी समस्या से संपर्क करना शुरू कर दें।

इस लैब की जाँच निम्नानुसार आयोजित की जाती है: हम प्रतिभागी के काफ़्का को एक नियंत्रण डेटा पैकेट भेजते हैं, फिर गोबलिन डेटा पैकेट को एचडीएफएस में स्थानांतरित करता है, फिर एयरफ़्लो इस डेटा पैकेट को लेता है और इसे क्लिकहाउस में डालता है। चाल यह है कि एयरफ्लो को वास्तविक समय में ऐसा नहीं करना चाहिए, यह इसे एक समय पर करता है: 15 मिनट में एक बार यह फाइलों का एक गुच्छा लेता है और इसे अंदर फेंकता है।

यह पता चला है कि हमें यहां और अब के चेकर के अनुरोध पर किसी तरह अपने स्वयं के डीएजी को ट्रिगर करने की आवश्यकता है। Googling, हमने पाया कि Airflow के बाद के संस्करणों के लिए तथाकथित प्रायोगिक एपीआई है । शब्द experimental , बेशक, डरावना लगता है, लेकिन क्या करना है ... अचानक यह उड़ जाएगा।

अगला, हम पूरे तरीके का वर्णन करते हैं: एयरफ्लो स्थापित करने से लेकर POST अनुरोध उत्पन्न करने के लिए जो प्रायोगिक एपीआई का उपयोग करके DAG से चलाता है। हम Ubuntu 16.04 के साथ काम करेंगे।

1. Airflow स्थापित करना

आइए देखें कि हमारे पास पायथन 3 और virtualenv है।

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

यदि इनमें से कोई भी गायब है, तो इंस्टॉल करें।

अब एक निर्देशिका बनाएं जिसमें हम एयरफ्लो के साथ काम करना जारी रखेंगे।

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

Airflow स्थापित करें:

 (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.cfg बनाई:

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

एयरफ्लो में एक वेब इंटरफेस है। यह कमांड चलाकर लॉन्च किया जा सकता है:

 (venv) $ airflow webserver --port 8081 

अब आप होस्ट 80 पोर्ट पर एक ब्राउज़र में वेब इंटरफ़ेस का उपयोग कर सकते हैं, जहां एयरफ्लो लॉन्च किया गया था, उदाहरण के लिए: <hostname:8081> ।

2. प्रायोगिक एपीआई के साथ काम करना

इस पर, Airflow कॉन्फ़िगर किया गया है और जाने के लिए तैयार है। हालाँकि, हमें प्रायोगिक एपीआई को भी चलाने की आवश्यकता है। हमारे चेकर्स पायथन में लिखे गए हैं, इसलिए आगे के सभी अनुरोध requests पुस्तकालय का उपयोग करते हुए उस पर होंगे।

वास्तव में, एपीआई पहले से ही सरल प्रश्नों के लिए काम कर रहा है। उदाहरण के लिए, ऐसा अनुरोध आपको इसके संचालन का परीक्षण करने की अनुमति देता है:

 >>> 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() 

फिर, आपको सामान्य अधिकारों के साथ एक उपयोगकर्ता बनाने की आवश्यकता है, जिन्हें डीएजी ट्रिगर बनाने की अनुमति होगी।

 >>> 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 अनुरोध स्वयं इस तरह दिखाई देगा:

 >>> 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 तालिका से अनुरोध करते हैं, डेटा के नियंत्रण पैकेट को पकड़ने की कोशिश कर रहे हैं।

सत्यापन पूरा हुआ।