പരീക്ഷണാത്മക API ഉപയോഗിച്ച് എയർഫ്ലോയിൽ ഒരു DAG ട്രിഗർ എങ്ങനെ നിർമ്മിക്കാം

ഞങ്ങളുടെ വിദ്യാഭ്യാസ പരിപാടികൾ തയ്യാറാക്കുന്നതിൽ, ചില ഉപകരണങ്ങളുമായി പ്രവർത്തിക്കുമ്പോൾ ഞങ്ങൾ ഇടയ്ക്കിടെ ബുദ്ധിമുട്ടുകൾ നേരിടുന്നു. ഞങ്ങൾ അവരെ അഭിമുഖീകരിക്കുന്ന നിമിഷത്തിൽ, ഈ പ്രശ്നത്തെ നേരിടാൻ സഹായിക്കുന്ന മതിയായ ഡോക്യുമെന്റേഷനുകളും ലേഖനങ്ങളും എല്ലായ്പ്പോഴും ഇല്ല.

ഉദാഹരണത്തിന്, 2015-ൽ, ബിഗ് ഡാറ്റാ സ്പെഷ്യലിസ്റ്റ് പ്രോഗ്രാമിൽ ഒരേസമയം 35 ഉപയോക്താക്കൾക്കായി ഞങ്ങൾ സ്പാർക്കിനൊപ്പം ഹഡൂപ്പ് ക്ലസ്റ്റർ ഉപയോഗിച്ചു. YARN ഉപയോഗിച്ച് അത്തരമൊരു ഉപയോക്തൃ കേസിനായി ഇത് എങ്ങനെ തയ്യാറാക്കണമെന്ന് വ്യക്തമല്ല. തൽഫലമായി, അവർ സ്വന്തമായി പാത കണ്ടെത്തി, നടന്നു ഹബ്രെയിലെ പോസ്റ്റ് കൂടാതെ അവതരിപ്പിച്ചു മോസ്കോ സ്പാർക്ക് മീറ്റപ്പ്.

ചരിത്രാതീതകാലം

ഇത്തവണ നമ്മൾ സംസാരിക്കുന്നത് മറ്റൊരു പരിപാടിയെ കുറിച്ചാണ് - ഡാറ്റ എഞ്ചിനീയർ. ഞങ്ങളുടെ പങ്കാളികൾ അതിൽ രണ്ട് തരം വാസ്തുവിദ്യകൾ നിർമ്മിക്കുന്നു: ലാംഡയും കപ്പയും. ലാംഡ്ബ ആർക്കിടെക്ചറിൽ, എച്ച്ഡിഎഫ്എസിൽ നിന്ന് ക്ലിക്ക്ഹൗസിലേക്ക് ലോഗുകൾ കൈമാറുന്നതിന് ബാച്ച് പ്രോസസ്സിംഗിന്റെ ഭാഗമായി എയർഫ്ലോ ഉപയോഗിക്കുന്നു.

എല്ലാം പൊതുവെ നല്ലതാണ്. അവർ പൈപ്പ് ലൈനുകൾ നിർമ്മിക്കട്ടെ. എന്നിരുന്നാലും, ഒരു "പക്ഷേ" ഉണ്ട്: ഞങ്ങളുടെ എല്ലാ പ്രോഗ്രാമുകളും പഠന പ്രക്രിയയുടെ കാര്യത്തിൽ സാങ്കേതികമായി പുരോഗമിച്ചവയാണ്. ലാബ് പരിശോധിക്കാൻ, ഞങ്ങൾ ഓട്ടോമാറ്റിക് ചെക്കറുകൾ ഉപയോഗിക്കുന്നു: പങ്കെടുക്കുന്നയാൾ തന്റെ സ്വകാര്യ അക്കൗണ്ടിലേക്ക് പോകേണ്ടതുണ്ട്, "ചെക്ക്" ബട്ടണിൽ ക്ലിക്ക് ചെയ്യുക, കുറച്ച് സമയത്തിന് ശേഷം അവൻ ചെയ്തതിനെക്കുറിച്ചുള്ള ചില വിപുലീകൃത ഫീഡ്ബാക്ക് അവൻ കാണുന്നു. ഈ ഘട്ടത്തിലാണ് നമ്മൾ നമ്മുടെ പ്രശ്നത്തെ സമീപിക്കാൻ തുടങ്ങുന്നത്.

ഈ ലാബ് പരിശോധിക്കുന്നത് ഇനിപ്പറയുന്ന രീതിയിൽ ക്രമീകരിച്ചിരിക്കുന്നു: പങ്കാളിയുടെ കാഫ്കയിലേക്ക് ഞങ്ങൾ ഒരു നിയന്ത്രണ ഡാറ്റ പാക്കറ്റ് അയയ്ക്കുന്നു, തുടർന്ന് Gobblin ഈ ഡാറ്റാ പാക്കറ്റ് HDFS-ലേക്ക് മാറ്റുന്നു, തുടർന്ന് Airflow ഈ ഡാറ്റാ പാക്കറ്റ് എടുത്ത് ClickHouse-ൽ ഇടുന്നു. എയർഫ്ലോ ഇത് തത്സമയം ചെയ്യേണ്ടതില്ല എന്നതാണ്, ഇത് ഷെഡ്യൂളിൽ ചെയ്യുന്നു: ഓരോ 15 മിനിറ്റിലും ഒരു കൂട്ടം ഫയലുകൾ എടുത്ത് അവ അപ്‌ലോഡ് ചെയ്യുന്നു.

ചെക്കർ ഇവിടെയും ഇപ്പോളും പ്രവർത്തിക്കുമ്പോൾ ഞങ്ങളുടെ അഭ്യർത്ഥന പ്രകാരം എങ്ങനെയെങ്കിലും അവരുടെ DAG ട്രിഗർ ചെയ്യേണ്ടതുണ്ടെന്ന് ഇത് മാറുന്നു. ഗൂഗ്ലിങ്ങിൽ, എയർഫ്ലോയുടെ പിന്നീടുള്ള പതിപ്പുകൾക്ക് വിളിക്കപ്പെടുന്ന ഒന്ന് ഉണ്ടെന്ന് ഞങ്ങൾ കണ്ടെത്തി പരീക്ഷണാത്മക API. വാക്ക് experimental, തീർച്ചയായും, ഇത് ഭയപ്പെടുത്തുന്നതായി തോന്നുന്നു, പക്ഷേ എന്തുചെയ്യണം ... അത് പെട്ടെന്ന് എടുക്കുന്നു.

അടുത്തതായി, ഞങ്ങൾ മുഴുവൻ പാതയും വിവരിക്കും: എയർഫ്ലോ ഇൻസ്റ്റാൾ ചെയ്യുന്നത് മുതൽ പരീക്ഷണാത്മക API ഉപയോഗിച്ച് ഒരു DAG ട്രിഗർ ചെയ്യുന്ന ഒരു POST അഭ്യർത്ഥന സൃഷ്ടിക്കുന്നത് വരെ. ഞങ്ങൾ ഉബുണ്ടു 16.04 ഉപയോഗിച്ച് പ്രവർത്തിക്കും.

1. എയർഫ്ലോ ഇൻസ്റ്റാളേഷൻ

നമുക്ക് Python 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) $

എയർഫ്ലോ ഇൻസ്റ്റാൾ ചെയ്യുക:

(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-ലെ ബ്രൗസറിൽ നിങ്ങൾക്ക് ഇപ്പോൾ വെബ് ഇന്റർഫേസ് ആക്‌സസ് ചെയ്യാൻ കഴിയും: <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