પ્રાયોગિક API નો ઉપયોગ કરીને એરફ્લોમાં DAG ટ્રિગર કેવી રીતે બનાવવું

અમારા શૈક્ષણિક કાર્યક્રમો તૈયાર કરતી વખતે, અમે સમયાંતરે અમુક સાધનો સાથે કામ કરવામાં મુશ્કેલીઓનો સામનો કરીએ છીએ. અને આ ક્ષણે જ્યારે આપણે તેમનો સામનો કરીએ છીએ, ત્યાં હંમેશા પૂરતા દસ્તાવેજો અને લેખો નથી કે જે અમને આ સમસ્યાનો સામનો કરવામાં મદદ કરે.

આ કેસ હતો, ઉદાહરણ તરીકે, 2015 માં, અને “બિગ ડેટા સ્પેશિયાલિસ્ટ” પ્રોગ્રામ દરમિયાન અમે એક સાથે 35 વપરાશકર્તાઓ માટે સ્પાર્ક સાથેના હડુપ ક્લસ્ટરનો ઉપયોગ કર્યો હતો. યાર્નનો ઉપયોગ કરીને આવા ઉપયોગ કેસ માટે તેને કેવી રીતે તૈયાર કરવું તે સ્પષ્ટ ન હતું. અંતે, તે શોધી કાઢ્યા અને અમારી જાતે જ માર્ગ પર ચાલ્યા, અમે કર્યું Habré પર પોસ્ટ અને ખાતે પણ પ્રદર્શન કર્યું હતું મોસ્કો સ્પાર્ક મીટઅપ.

પ્રાગૈતિહાસિક

આ વખતે આપણે એક અલગ પ્રોગ્રામ વિશે વાત કરીશું - ડેટા ઇજનેર. અમારા સહભાગીઓ તેના પર બે પ્રકારના આર્કિટેક્ચર બનાવે છે: લેમ્બડા અને કપ્પા. અને lamdba આર્કિટેક્ચરમાં, બેચ પ્રોસેસિંગના ભાગ રૂપે, એરફ્લોનો ઉપયોગ HDFS થી ક્લિકહાઉસમાં લોગ ટ્રાન્સફર કરવા માટે થાય છે.

બધું સામાન્ય રીતે સારું છે. તેમને પોતાની પાઇપલાઇન બનાવવા દો. જો કે, ત્યાં એક "પરંતુ" છે: અમારા બધા પ્રોગ્રામ્સ શીખવાની પ્રક્રિયાના દૃષ્ટિકોણથી તકનીકી રીતે અદ્યતન છે. લેબ તપાસવા માટે, અમે સ્વચાલિત ચેકર્સનો ઉપયોગ કરીએ છીએ: સહભાગીને તેના વ્યક્તિગત ખાતામાં જવાની જરૂર છે, "ચેક" બટનને ક્લિક કરો, અને થોડા સમય પછી તેણે જે કર્યું તેના પર કોઈ પ્રકારનો વિસ્તૃત પ્રતિસાદ જુએ છે. અને તે આ ક્ષણે છે કે આપણે આપણી સમસ્યાનો સંપર્ક કરવાનું શરૂ કરીએ છીએ.

આ લેબની ચકાસણી આ રીતે રચાયેલ છે: અમે સહભાગીના કાફ્કાને કંટ્રોલ ડેટા પેકેટ મોકલીએ છીએ, પછી ગોબ્લિન આ ડેટા પેકેટને HDFS પર ટ્રાન્સફર કરે છે, પછી એરફ્લો આ ડેટા પેકેટ લે છે અને તેને ક્લિકહાઉસમાં મૂકે છે. યુક્તિ એ છે કે એરફ્લોને વાસ્તવિક સમયમાં આ કરવાની જરૂર નથી, તે શેડ્યૂલ અનુસાર કરે છે: દર 15 મિનિટે તે ફાઇલોનો સમૂહ લે છે અને તેને અપલોડ કરે છે.

તે તારણ આપે છે કે અમારે અમારી વિનંતી પર કોઈક રીતે તેમના DAG ને ટ્રિગર કરવાની જરૂર છે જ્યારે ચેકર અહીં અને અત્યારે ચાલી રહ્યું છે. ગૂગલિંગ કર્યા પછી, અમને જાણવા મળ્યું કે એરફ્લોના પછીના સંસ્કરણો માટે કહેવાતા છે પ્રાયોગિક API. શબ્દ experimental, અલબત્ત, તે ડરામણી લાગે છે, પરંતુ શું કરવું... અચાનક તે બંધ થઈ જાય છે.

આગળ, અમે સમગ્ર પાથનું વર્ણન કરીશું: એરફ્લો ઇન્સ્ટોલ કરવાથી લઈને POST વિનંતી જનરેટ કરવા સુધી જે પ્રાયોગિક API નો ઉપયોગ કરીને DAG ને ટ્રિગર કરે છે. અમે ઉબુન્ટુ 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