როგორ გავაკეთოთ DAG ტრიგერი Airflow-ში ექსპერიმენტული API-ის გამოყენებით

ჩვენი საგანმანათლებლო პროგრამების მომზადებისას პერიოდულად ვაწყდებით სირთულეებს გარკვეულ ინსტრუმენტებთან მუშაობის კუთხით. და იმ მომენტში, როდესაც მათ ვხვდებით, ყოველთვის არ არის საკმარისი დოკუმენტაცია და სტატიები, რომლებიც დაგეხმარებათ ამ პრობლემის მოგვარებაში.

ასე იყო, მაგალითად, 2015 წელს და ჩვენ გამოვიყენეთ Hadoop კლასტერი Spark-თან ერთად 35 ერთდროული მომხმარებლისთვის Big Data Specialist პროგრამაზე. გაუგებარი იყო, როგორ უნდა მომზადდეს ის ასეთი მომხმარებლის შემთხვევისთვის YARN-ის გამოყენებით. შედეგად, მათ გაარკვიეს და საკუთარი გზით გაიარეს გზა პოსტი Habré-ზე და ასევე შეასრულა მოსკოვის ნაპერწკლის შეხვედრა.

წინაისტორია

ამჯერად განსხვავებულ პროგრამაზე ვისაუბრებთ - მონაცემთა ინჟინერი. მასზე ჩვენი მონაწილეები აშენებენ ორი ტიპის არქიტექტურას: ლამბდას და კაპას. და lamdba-ს არქიტექტურაში, Airflow გამოიყენება, როგორც სერიული დამუშავების ნაწილი, ჟურნალების გადასატანად HDFS-დან ClickHouse-ში.

ზოგადად ყველაფერი კარგია. დაე, ააშენონ თავიანთი მილსადენები. თუმცა, არის ერთი „მაგრამ“: ყველა ჩვენი პროგრამა ტექნოლოგიურად განვითარებულია თავად სასწავლო პროცესის თვალსაზრისით. ლაბორატორიის შესამოწმებლად, ჩვენ ვიყენებთ ავტომატურ ქამრებს: მონაწილეს უნდა გადავიდეს მის პირად ანგარიშზე, დააწკაპუნეთ ღილაკზე „შემოწმება“ და გარკვეული პერიოდის შემდეგ ის ხედავს ერთგვარ გაფართოებულ გამოხმაურებას იმის შესახებ, რაც გააკეთა. და სწორედ ამ დროს ვიწყებთ ჩვენს პრობლემას.

ამ ლაბორატორიის შემოწმება ხდება შემდეგნაირად: ჩვენ ვუგზავნით საკონტროლო მონაცემთა პაკეტს მონაწილის კაფკას, შემდეგ Gobblin გადასცემს ამ მონაცემთა პაკეტს HDFS-ში, შემდეგ Airflow იღებს ამ მონაცემთა პაკეტს და დებს მას ClickHouse-ში. ხრიკი იმაშია, რომ Airflow-ს არ სჭირდება ამის გაკეთება რეალურ დროში, ის ამას აკეთებს გრაფიკით: ყოველ 15 წუთში ერთხელ სჭირდება ფაილების თაიგულს და ატვირთავს მათ.

გამოდის, რომ ჩვენ გვჭირდება როგორმე გავააქტიუროთ მათი DAG ჩვენი მოთხოვნით, სანამ შემოწმება მუშაობს აქ და ახლა. გუგლში გავარკვიეთ, რომ 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 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-სთან მუშაობა

ჰაერის ნაკადი კონფიგურირებულია და მზად არის წასასვლელად. თუმცა, ჩვენ ასევე გვჭირდება ექსპერიმენტული 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-ს გარკვეული დროის დასამუშავებლად და მოთხოვნის გაგზავნას ClickHouse ცხრილზე, ვცდილობთ დაიჭიროთ საკონტროლო მონაცემთა პაკეტი.

დადასტურება დასრულდა.

წყარო: www.habr.com