របៀបបង្កើត DAG trigger នៅក្នុង Airflow ដោយប្រើ API សាកល្បង

ក្នុងការរៀបចំកម្មវិធីអប់រំរបស់យើង យើងជួបប្រទះការលំបាកជាប្រចាំទាក់ទងនឹងការធ្វើការជាមួយឧបករណ៍មួយចំនួន។ ហើយនៅពេលនេះនៅពេលដែលយើងជួបប្រទះពួកគេ វាមិនតែងតែមានឯកសារ និងអត្ថបទគ្រប់គ្រាន់ដែលអាចជួយដោះស្រាយបញ្ហានេះបានទេ។

ដូច្នេះវាជាឧទាហរណ៍ក្នុងឆ្នាំ 2015 ហើយយើងបានប្រើចង្កោម Hadoop ជាមួយ Spark សម្រាប់អ្នកប្រើប្រាស់ 35 នាក់ក្នុងពេលដំណាលគ្នានៅលើកម្មវិធី Big Data Specialist ។ វាមិនច្បាស់អំពីរបៀបរៀបចំវាសម្រាប់ករណីអ្នកប្រើប្រាស់បែបនេះដោយប្រើ YARN ទេ។ ជាលទ្ធផល ដោយបានស្វែងយល់ និងដើរលើផ្លូវដោយខ្លួនឯង ពួកគេបានធ្វើ ប្រកាសនៅលើ Habre និងបានអនុវត្តផងដែរ។ កិច្ចប្រជុំ Spark ទីក្រុងម៉ូស្គូ.

សម័យបុរេប្រវត្តិសាស្ត្រ

លើកនេះយើងនឹងនិយាយអំពីកម្មវិធីផ្សេង - វិស្វករទិន្នន័យ. នៅលើវា អ្នកចូលរួមរបស់យើងសាងសង់ស្ថាបត្យកម្មពីរប្រភេទគឺ lambda និង kappa ។ ហើយនៅក្នុងស្ថាបត្យកម្ម lamdba Airflow ត្រូវបានប្រើជាផ្នែកនៃដំណើរការបណ្តុំដើម្បីផ្ទេរកំណត់ហេតុពី HDFS ទៅ ClickHouse ។

ជាទូទៅអ្វីៗគឺល្អ។ អនុញ្ញាតឱ្យពួកគេសាងសង់បំពង់របស់ពួកគេ។ ទោះយ៉ាងណាក៏ដោយ មាន "ប៉ុន្តែ"៖ កម្មវិធីរបស់យើងទាំងអស់គឺមានភាពជឿនលឿនខាងបច្ចេកវិទ្យាទាក់ទងនឹងដំណើរការសិក្សាដោយខ្លួនឯង។ ដើម្បីពិនិត្យមើលមន្ទីរពិសោធន៍ យើងប្រើឧបករណ៍ពិនិត្យដោយស្វ័យប្រវត្តិ៖ អ្នកចូលរួមត្រូវចូលទៅកាន់គណនីផ្ទាល់ខ្លួនរបស់គាត់ ចុចប៊ូតុង "ពិនិត្យ" ហើយមួយសន្ទុះក្រោយមកគាត់ឃើញប្រភេទនៃមតិកែលម្អមួយចំនួនលើអ្វីដែលគាត់បានធ្វើ។ ហើយវាគឺនៅចំណុចនេះដែលយើងចាប់ផ្តើមខិតទៅរកបញ្ហារបស់យើង។

ការពិនិត្យមើលមន្ទីរពិសោធន៍នេះត្រូវបានរៀបចំដូចខាងក្រោម៖ យើងផ្ញើកញ្ចប់ទិន្នន័យត្រួតពិនិត្យទៅ Kafka របស់អ្នកចូលរួម បន្ទាប់មក Gobblin ផ្ទេរកញ្ចប់ទិន្នន័យនេះទៅ HDFS បន្ទាប់មក Airflow យកកញ្ចប់ទិន្នន័យនេះ ហើយដាក់វានៅក្នុង ClickHouse ។ ល្បិចគឺថា Airflow មិនចាំបាច់ធ្វើបែបនេះក្នុងពេលវេលាជាក់ស្តែងនោះទេ វាធ្វើវាតាមកាលវិភាគ៖ រៀងរាល់ 15 នាទីម្តង វាត្រូវការឯកសារជាច្រើន ហើយបង្ហោះវា។

វាប្រែថាយើងចាំបាច់ត្រូវចាប់ផ្តើម DAG របស់ពួកគេដោយខ្លួនឯងតាមសំណើរបស់យើងខណៈពេលដែលអ្នកត្រួតពិនិត្យកំពុងដំណើរការនៅទីនេះនិងឥឡូវនេះ។ Googling យើងបានរកឃើញថាសម្រាប់កំណែក្រោយនៃ Airflow គឺជាអ្វីដែលគេហៅថា API ពិសោធន៍។ ពាក្យ experimentalពិតណាស់ វាស្តាប់ទៅគួរឱ្យខ្លាច ប៉ុន្តែអ្វីដែលត្រូវធ្វើ ... វាស្រាប់តែរលត់ទៅវិញ។

បន្ទាប់ យើងនឹងរៀបរាប់អំពីផ្លូវទាំងមូល៖ ពីការដំឡើង Airflow ដល់ការបង្កើតសំណើ POST ដែលបង្កឱ្យមាន DAG ដោយប្រើ API ពិសោធន៍។ យើងនឹងធ្វើការជាមួយ 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 បានបង្កើតឯកសារកំណត់រចនាសម្ព័ន្ធផ្ទាល់ខ្លួនរបស់វា។ airflow.cfg в AIRFLOW_HOME:

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

Airflow មានចំណុចប្រទាក់បណ្ដាញ។ វាអាចត្រូវបានបើកដំណើរការដោយដំណើរការពាក្យបញ្ជា:

(venv) $ airflow webserver --port 8081

ឥឡូវនេះអ្នកអាចចូលប្រើចំណុចប្រទាក់បណ្ដាញនៅក្នុងកម្មវិធីរុករកតាមច្រក 8081 នៅលើម៉ាស៊ីនដែល Airflow កំពុងដំណើរការដូចនេះ៖ <hostname:8081>.

2. ធ្វើការជាមួយ API ពិសោធន៍

នៅលើ Airflow នេះត្រូវបានកំណត់រចនាសម្ព័ន្ធ និងត្រៀមរួចរាល់។ ទោះយ៉ាងណាក៏ដោយ យើងក៏ត្រូវដំណើរការ 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 នោះ យើងដំណើរការទៅការពិតដែលថាសំណើប្រភេទនេះមិនអាចត្រូវបានធ្វើឡើងដោយគ្មានការផ្ទៀងផ្ទាត់នោះទេ។

ដើម្បីធ្វើដូចនេះអ្នកនឹងត្រូវធ្វើសកម្មភាពមួយចំនួន។

ដំបូងអ្នកត្រូវបន្ថែមវាទៅ config៖

[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 trigger ។

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