प्रयोगात्मक API प्रयोग गरेर एयरफ्लोमा DAG ट्रिगर कसरी बनाउने

हाम्रा शैक्षिक कार्यक्रमहरू तयार गर्दा, हामीले आवधिक रूपमा केही उपकरणहरूसँग काम गर्ने सन्दर्भमा कठिनाइहरूको सामना गर्छौं। र यस क्षणमा जब हामी तिनीहरूसँग भेट्छौं, त्यहाँ सधैं पर्याप्त कागजातहरू र लेखहरू छैनन् जुन यस समस्याको सामना गर्न मद्दत गर्दछ।

त्यसोभए यो, उदाहरणका लागि, 2015 मा थियो, र हामीले बिग डाटा विशेषज्ञ कार्यक्रममा 35 एक साथ प्रयोगकर्ताहरूको लागि स्पार्कको साथ Hadoop क्लस्टर प्रयोग गर्यौं। यो यार्न प्रयोग गरेर यस्तो प्रयोगकर्ता केसको लागि कसरी तयार गर्ने भन्ने स्पष्ट थिएन। नतिजाको रूपमा, तिनीहरूले आफैंले बाटो पत्ता लगाएर हिँडे Habré मा पोस्ट र प्रदर्शन पनि गरे मस्को स्पार्क बैठक.

प्रागितिहास

यस पटक हामी फरक कार्यक्रमको बारेमा कुरा गर्नेछौं - डाटा ईन्जिनियर। यसमा, हाम्रा सहभागीहरूले दुई प्रकारको वास्तुकला निर्माण गर्छन्: लाम्ब्डा र कापा। र lamdba आर्किटेक्चरमा, Airflow लाई HDFS बाट ClickHouse मा लगहरू स्थानान्तरण गर्न ब्याच प्रशोधनको भागको रूपमा प्रयोग गरिन्छ।

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

यो ल्याब जाँच गर्दै निम्नानुसार व्यवस्थित गरिएको छ: हामी सहभागीको काफ्कालाई नियन्त्रण डेटा प्याकेट पठाउँछौं, त्यसपछि Gobblin ले यो डेटा प्याकेट HDFS मा स्थानान्तरण गर्छ, त्यसपछि Airflow ले यो डेटा प्याकेट लिन्छ र यसलाई ClickHouse मा राख्छ। चाल यो हो कि Airflow ले यो वास्तविक समयमा गर्नु पर्दैन, यो समय तालिकामा गर्छ: एक पटक प्रत्येक 15 मिनेटमा यसले फाइलहरूको गुच्छा लिन्छ र तिनीहरूलाई अपलोड गर्दछ।

यो बाहिर जान्छ कि हामीले कुनै न कुनै रूपमा हाम्रो अनुरोधमा तिनीहरूको DAG आफैं ट्रिगर गर्न आवश्यक छ जब परीक्षक यहाँ र अहिले चलिरहेको छ। गुगलिङ, हामीले पत्ता लगायौं कि एयरफ्लोको पछिल्ला संस्करणहरूको लागि त्यहाँ तथाकथित छ प्रयोगात्मक API। शब्द experimental, अवश्य पनि, यो डरलाग्दो सुनिन्छ, तर के गर्ने ... यो अचानक बन्द हुन्छ।

अर्को, हामी सम्पूर्ण मार्गको वर्णन गर्नेछौं: एयरफ्लो स्थापना गर्नेदेखि लिएर प्रायोगिक API प्रयोग गरेर DAG ट्रिगर गर्ने POST अनुरोध उत्पन्न गर्ने। हामी Ubuntu 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.cfg в AIRFLOW_HOME:

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

Airflow सँग वेब इन्टरफेस छ। यो आदेश चलाएर सुरू गर्न सकिन्छ:

(venv) $ airflow webserver --port 8081

तपाईले अब ब्राउजरमा पोर्ट 8081 मा होस्टमा वेब इन्टरफेस पहुँच गर्न सक्नुहुन्छ जहाँ एयरफ्लो चलिरहेको थियो, जस्तै: <hostname:8081>.

२. प्रयोगात्मक 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