ਪ੍ਰਯੋਗਾਤਮਕ API ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਏਅਰਫਲੋ ਵਿੱਚ ਇੱਕ DAG ਟਰਿੱਗਰ ਕਿਵੇਂ ਬਣਾਇਆ ਜਾਵੇ

ਸਾਡੇ ਵਿਦਿਅਕ ਪ੍ਰੋਗਰਾਮਾਂ ਨੂੰ ਤਿਆਰ ਕਰਦੇ ਸਮੇਂ, ਸਾਨੂੰ ਸਮੇਂ-ਸਮੇਂ 'ਤੇ ਕੁਝ ਸਾਧਨਾਂ ਨਾਲ ਕੰਮ ਕਰਨ ਦੇ ਮਾਮਲੇ ਵਿੱਚ ਮੁਸ਼ਕਲਾਂ ਦਾ ਸਾਹਮਣਾ ਕਰਨਾ ਪੈਂਦਾ ਹੈ। ਅਤੇ ਇਸ ਸਮੇਂ ਜਦੋਂ ਅਸੀਂ ਉਹਨਾਂ ਦਾ ਸਾਹਮਣਾ ਕਰਦੇ ਹਾਂ, ਇੱਥੇ ਹਮੇਸ਼ਾ ਲੋੜੀਂਦੇ ਦਸਤਾਵੇਜ਼ ਅਤੇ ਲੇਖ ਨਹੀਂ ਹੁੰਦੇ ਹਨ ਜੋ ਇਸ ਸਮੱਸਿਆ ਨਾਲ ਸਿੱਝਣ ਵਿੱਚ ਸਾਡੀ ਮਦਦ ਕਰਨਗੇ।

ਇਹ ਮਾਮਲਾ ਸੀ, ਉਦਾਹਰਨ ਲਈ, 2015 ਵਿੱਚ, ਅਤੇ "ਬਿਗ ਡੇਟਾ ਸਪੈਸ਼ਲਿਸਟ" ਪ੍ਰੋਗਰਾਮ ਦੇ ਦੌਰਾਨ ਅਸੀਂ 35 ਇੱਕੋ ਸਮੇਂ ਦੇ ਉਪਭੋਗਤਾਵਾਂ ਲਈ ਸਪਾਰਕ ਦੇ ਨਾਲ ਇੱਕ ਹੈਡੂਪ ਕਲੱਸਟਰ ਦੀ ਵਰਤੋਂ ਕੀਤੀ। ਇਹ ਸਪੱਸ਼ਟ ਨਹੀਂ ਸੀ ਕਿ ਯਾਰਨ ਦੀ ਵਰਤੋਂ ਕਰਕੇ ਇਸ ਨੂੰ ਅਜਿਹੇ ਵਰਤੋਂ ਦੇ ਕੇਸ ਲਈ ਕਿਵੇਂ ਤਿਆਰ ਕਰਨਾ ਹੈ। ਅੰਤ ਵਿੱਚ, ਇਸਦਾ ਪਤਾ ਲਗਾ ਕੇ ਅਤੇ ਆਪਣੇ ਆਪ ਹੀ ਰਸਤੇ ਤੇ ਚੱਲਦੇ ਹੋਏ, ਅਸੀਂ ਕੀਤਾ Habré 'ਤੇ ਪੋਸਟ ਅਤੇ ਵਿਖੇ ਵੀ ਪ੍ਰਦਰਸ਼ਨ ਕੀਤਾ ਮਾਸਕੋ ਸਪਾਰਕ ਮੀਟਿੰਗ.

prehistory

ਇਸ ਵਾਰ ਅਸੀਂ ਇੱਕ ਵੱਖਰੇ ਪ੍ਰੋਗਰਾਮ ਬਾਰੇ ਗੱਲ ਕਰਾਂਗੇ- ਡਾਟਾ ਇੰਜੀਨੀਅਰ. ਸਾਡੇ ਭਾਗੀਦਾਰ ਇਸ 'ਤੇ ਦੋ ਕਿਸਮ ਦੇ ਆਰਕੀਟੈਕਚਰ ਬਣਾਉਂਦੇ ਹਨ: ਲਾਂਬਡਾ ਅਤੇ ਕਪਾ। ਅਤੇ lamdba ਆਰਕੀਟੈਕਚਰ ਵਿੱਚ, ਬੈਚ ਪ੍ਰੋਸੈਸਿੰਗ ਦੇ ਹਿੱਸੇ ਵਜੋਂ, ਏਅਰਫਲੋ ਦੀ ਵਰਤੋਂ ਐਚਡੀਐਫਐਸ ਤੋਂ ਕਲਿਕਹਾਊਸ ਵਿੱਚ ਲਾਗਾਂ ਨੂੰ ਟ੍ਰਾਂਸਫਰ ਕਰਨ ਲਈ ਕੀਤੀ ਜਾਂਦੀ ਹੈ।

ਸਭ ਕੁਝ ਆਮ ਤੌਰ 'ਤੇ ਚੰਗਾ ਹੈ. ਉਨ੍ਹਾਂ ਨੂੰ ਆਪਣੀਆਂ ਪਾਈਪਲਾਈਨਾਂ ਬਣਾਉਣ ਦਿਓ। ਹਾਲਾਂਕਿ, ਇੱਥੇ ਇੱਕ "ਪਰ" ਹੈ: ਸਾਡੇ ਸਾਰੇ ਪ੍ਰੋਗਰਾਮ ਸਿੱਖਣ ਦੀ ਪ੍ਰਕਿਰਿਆ ਦੇ ਦ੍ਰਿਸ਼ਟੀਕੋਣ ਤੋਂ ਤਕਨੀਕੀ ਤੌਰ 'ਤੇ ਉੱਨਤ ਹਨ। ਪ੍ਰਯੋਗਸ਼ਾਲਾ ਦੀ ਜਾਂਚ ਕਰਨ ਲਈ, ਅਸੀਂ ਆਟੋਮੈਟਿਕ ਚੈਕਰਾਂ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹਾਂ: ਭਾਗੀਦਾਰ ਨੂੰ ਆਪਣੇ ਨਿੱਜੀ ਖਾਤੇ 'ਤੇ ਜਾਣ ਦੀ ਲੋੜ ਹੁੰਦੀ ਹੈ, "ਚੈੱਕ" ਬਟਨ 'ਤੇ ਕਲਿੱਕ ਕਰੋ, ਅਤੇ ਕੁਝ ਸਮੇਂ ਬਾਅਦ ਉਸ ਨੇ ਜੋ ਕੀਤਾ ਉਸ 'ਤੇ ਕਿਸੇ ਕਿਸਮ ਦਾ ਵਿਸਤ੍ਰਿਤ ਫੀਡਬੈਕ ਦੇਖਦਾ ਹੈ। ਅਤੇ ਇਹ ਇਸ ਪਲ 'ਤੇ ਹੈ ਕਿ ਅਸੀਂ ਆਪਣੀ ਸਮੱਸਿਆ ਵੱਲ ਜਾਣਾ ਸ਼ੁਰੂ ਕਰਦੇ ਹਾਂ.

ਇਸ ਲੈਬ ਦੀ ਤਸਦੀਕ ਇਸ ਤਰ੍ਹਾਂ ਕੀਤੀ ਗਈ ਹੈ: ਅਸੀਂ ਭਾਗੀਦਾਰ ਦੇ ਕਾਫਕਾ ਨੂੰ ਇੱਕ ਨਿਯੰਤਰਣ ਡੇਟਾ ਪੈਕੇਟ ਭੇਜਦੇ ਹਾਂ, ਫਿਰ ਗੌਬਲਿਨ ਇਸ ਡੇਟਾ ਪੈਕੇਟ ਨੂੰ HDFS ਵਿੱਚ ਟ੍ਰਾਂਸਫਰ ਕਰਦਾ ਹੈ, ਫਿਰ ਏਅਰਫਲੋ ਇਸ ਡੇਟਾ ਪੈਕੇਟ ਨੂੰ ਲੈਂਦਾ ਹੈ ਅਤੇ ਇਸਨੂੰ ਕਲਿਕਹਾਊਸ ਵਿੱਚ ਰੱਖਦਾ ਹੈ। ਚਾਲ ਇਹ ਹੈ ਕਿ ਏਅਰਫਲੋ ਨੂੰ ਇਹ ਅਸਲ ਸਮੇਂ ਵਿੱਚ ਨਹੀਂ ਕਰਨਾ ਪੈਂਦਾ, ਇਹ ਇੱਕ ਅਨੁਸੂਚੀ ਦੇ ਅਨੁਸਾਰ ਕਰਦਾ ਹੈ: ਹਰ 15 ਮਿੰਟ ਵਿੱਚ ਇਹ ਫਾਈਲਾਂ ਦਾ ਇੱਕ ਸਮੂਹ ਲੈਂਦਾ ਹੈ ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਅਪਲੋਡ ਕਰਦਾ ਹੈ।

ਇਹ ਪਤਾ ਚਲਦਾ ਹੈ ਕਿ ਸਾਨੂੰ ਆਪਣੀ ਬੇਨਤੀ 'ਤੇ ਉਨ੍ਹਾਂ ਦੇ ਡੀਏਜੀ ਨੂੰ ਆਪਣੇ ਆਪ ਟਰਿੱਗਰ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੈ ਜਦੋਂ ਕਿ ਚੈਕਰ ਇੱਥੇ ਅਤੇ ਹੁਣ ਚੱਲ ਰਿਹਾ ਹੈ। ਗੂਗਲਿੰਗ ਤੋਂ ਬਾਅਦ, ਸਾਨੂੰ ਪਤਾ ਲੱਗਾ ਕਿ ਏਅਰਫਲੋ ਦੇ ਬਾਅਦ ਦੇ ਸੰਸਕਰਣਾਂ ਲਈ ਇੱਕ ਅਖੌਤੀ ਹੈ ਪ੍ਰਯੋਗਾਤਮਕ 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'

ਬੇਨਤੀ 'ਤੇ ਸਫਲਤਾਪੂਰਵਕ ਪ੍ਰਕਿਰਿਆ ਕੀਤੀ ਗਈ ਸੀ।

ਇਸ ਅਨੁਸਾਰ, ਅਸੀਂ ਫਿਰ ਡੀਏਜੀ ਨੂੰ ਕਾਰਵਾਈ ਕਰਨ ਅਤੇ ਕਲਿਕਹਾਊਸ ਟੇਬਲ ਨੂੰ ਬੇਨਤੀ ਕਰਨ ਲਈ ਕੁਝ ਸਮਾਂ ਦਿੰਦੇ ਹਾਂ, ਕੰਟਰੋਲ ਡੇਟਾ ਪੈਕੇਟ ਨੂੰ ਫੜਨ ਦੀ ਕੋਸ਼ਿਸ਼ ਕਰਦੇ ਹੋਏ।

ਜਾਂਚ ਪੂਰੀ ਹੋਈ।

ਸਰੋਤ: www.habr.com