পরীক্ষামূলক API ব্যবহার করে এয়ারফ্লোতে কীভাবে একটি DAG ট্রিগার তৈরি করবেন

আমাদের শিক্ষামূলক প্রোগ্রামগুলি প্রস্তুত করার সময়, আমরা নির্দিষ্ট সরঞ্জামগুলির সাথে কাজ করার ক্ষেত্রে পর্যায়ক্রমে সমস্যার সম্মুখীন হই। এবং এই মুহুর্তে যখন আমরা তাদের মুখোমুখি হই, সেখানে সর্বদা যথেষ্ট ডকুমেন্টেশন এবং নিবন্ধ থাকে না যা আমাদের এই সমস্যাটি মোকাবেলা করতে সাহায্য করবে।

এটি ছিল, উদাহরণস্বরূপ, 2015 সালে, এবং "বিগ ডেটা স্পেশালিস্ট" প্রোগ্রামের সময় আমরা 35 জন একযোগে ব্যবহারকারীর জন্য স্পার্ক সহ একটি হ্যাডুপ ক্লাস্টার ব্যবহার করেছি। ইয়ার্ন ব্যবহার করে এই ধরনের ব্যবহারের ক্ষেত্রে কীভাবে এটি প্রস্তুত করা যায় তা পরিষ্কার ছিল না। শেষ পর্যন্ত, এটি খুঁজে বের করে এবং আমাদের নিজস্ব পথে হাঁটা, আমরা তা করেছি Habré উপর পোস্ট এবং এ সঞ্চালিত মস্কো স্পার্ক মিটআপ.

প্রাগঐতিহাসিক

এবার আমরা একটি ভিন্ন কর্মসূচি নিয়ে কথা বলব- ডেটা ইঞ্জিনিয়ার. আমাদের অংশগ্রহণকারীরা এটিতে দুটি ধরণের স্থাপত্য তৈরি করে: ল্যাম্বদা এবং কাপা। এবং lamdba আর্কিটেকচারে, ব্যাচ প্রক্রিয়াকরণের অংশ হিসাবে, এয়ারফ্লো HDFS থেকে ক্লিক হাউসে লগ স্থানান্তর করতে ব্যবহৃত হয়।

সবকিছুই সাধারণত ভালো। তাদের নিজস্ব পাইপলাইন তৈরি করতে দিন। যাইহোক, একটি কিন্তু আছে: আমাদের সমস্ত প্রোগ্রাম শেখার প্রক্রিয়ার দৃষ্টিকোণ থেকে প্রযুক্তিগতভাবে উন্নত। ল্যাব চেক করার জন্য, আমরা স্বয়ংক্রিয় চেকার ব্যবহার করি: অংশগ্রহণকারীকে তার ব্যক্তিগত অ্যাকাউন্টে যেতে হবে, "চেক" বোতামে ক্লিক করতে হবে এবং কিছু সময়ের পরে সে যা করেছে তার উপর কিছু বর্ধিত প্রতিক্রিয়া দেখতে পাবে। এবং এই মুহুর্তে আমরা আমাদের সমস্যার কাছে যেতে শুরু করি।

এই ল্যাবের যাচাইকরণটি এইভাবে গঠিত: আমরা অংশগ্রহণকারীর কাফকাকে একটি কন্ট্রোল ডেটা প্যাকেট পাঠাই, তারপর গবলিন এই ডেটা প্যাকেটটি HDFS-এ স্থানান্তর করে, তারপর Airflow এই ডেটা প্যাকেটটি নিয়ে ক্লিক হাউসে রাখে। কৌশলটি হল যে এয়ারফ্লোকে রিয়েল টাইমে এটি করতে হবে না, এটি একটি সময়সূচী অনুসারে এটি করে: প্রতি 15 মিনিটে এটি একগুচ্ছ ফাইল নেয় এবং সেগুলি আপলোড করে।

দেখা যাচ্ছে যে চেকার এখানে এবং এখন চলছে তখন আমাদের অনুরোধে তাদের ডিএজিকে নিজেদেরকে ট্রিগার করতে হবে। গুগলিংয়ের পরে, আমরা খুঁজে পেয়েছি যে এয়ারফ্লো এর পরবর্তী সংস্করণগুলির জন্য একটি তথাকথিত রয়েছে পরীক্ষামূলক API। শব্দ experimental, অবশ্যই, এটা ভীতিকর শোনাচ্ছে, কিন্তু কি করব... হঠাৎ করেই তা বন্ধ হয়ে যায়।

Далее опишем весь путь: от установки Airflow до формирования POST-запроса, который триггерит DAG, используя Experimental API. Работать будем с Ubuntu 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-এ একটি ব্রাউজারে ওয়েব ইন্টারফেসে আঘাত করতে পারেন যেখানে Airflow চলছিল, উদাহরণস্বরূপ: <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 অনুরোধ চালু করুন৷

পোস্ট অনুরোধ নিজেই এই মত দেখাবে:

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