ວິທີການສ້າງຜົນກະທົບຕໍ່ DAG ໃນ Airflow ໂດຍໃຊ້ API ທົດລອງ

ໃນເວລາທີ່ການກະກຽມໂຄງການການສຶກສາຂອງພວກເຮົາ, ພວກເຮົາແຕ່ລະໄລຍະພົບຄວາມຫຍຸ້ງຍາກໃນການເຮັດວຽກກັບເຄື່ອງມືສະເພາະໃດຫນຶ່ງ. ແລະໃນເວລານີ້ໃນເວລາທີ່ພວກເຮົາພົບພວກເຂົາ, ບໍ່ມີເອກະສານແລະບົດຄວາມພຽງພໍທີ່ຈະຊ່ວຍໃຫ້ພວກເຮົາຮັບມືກັບບັນຫານີ້.

ນີ້ແມ່ນກໍລະນີ, ຕົວຢ່າງ, ໃນປີ 2015, ແລະໃນລະຫວ່າງໂຄງການ "Big Data Specialist" ພວກເຮົາໄດ້ໃຊ້ກຸ່ມ Hadoop ກັບ Spark ສໍາລັບ 35 ຜູ້ໃຊ້ພ້ອມໆກັນ. ມັນບໍ່ຈະແຈ້ງກ່ຽວກັບວິທີການກະກຽມມັນສໍາລັບກໍລະນີການນໍາໃຊ້ດັ່ງກ່າວໂດຍໃຊ້ YARN. ໃນ​ທີ່​ສຸດ, ໄດ້​ຄິດ​ອອກ​ມັນ​ອອກ​ແລະ​ເດີນ​ທາງ​ຂອງ​ຕົນ​ເອງ, ພວກ​ເຮົາ​ໄດ້​ເຮັດ ໂພດໃນ Habre ແລະຍັງປະຕິບັດຢູ່ ມອດໂກ Spark Meetup.

ປະວັດສາດ

ເວລານີ້ພວກເຮົາຈະເວົ້າກ່ຽວກັບໂຄງການທີ່ແຕກຕ່າງກັນ - ນັກວິສະວະກອນຂໍ້ມູນ. ຜູ້ເຂົ້າຮ່ວມຂອງພວກເຮົາກໍ່ສ້າງສະຖາປັດຕະຍະກໍາສອງປະເພດຢູ່ເທິງມັນ: lambda ແລະ kappa. ແລະໃນສະຖາປັດຕະຍະກໍາ lamdba, ເປັນສ່ວນຫນຶ່ງຂອງການປຸງແຕ່ງ batch, Airflow ຖືກນໍາໃຊ້ເພື່ອໂອນບັນທຶກຈາກ HDFS ໄປ ClickHouse.

ທຸກສິ່ງທຸກຢ່າງໂດຍທົ່ວໄປແມ່ນດີ. ໃຫ້ພວກເຂົາສ້າງທໍ່ຂອງຕົນເອງ. ຢ່າງໃດກໍ່ຕາມ, ມີ "ແຕ່": ໂຄງການຂອງພວກເຮົາທັງຫມົດແມ່ນກ້າວຫນ້າທາງດ້ານເຕັກໂນໂລຢີຈາກທັດສະນະຂອງຂະບວນການຮຽນຮູ້ຕົວມັນເອງ. ເພື່ອກວດເບິ່ງຫ້ອງທົດລອງ, ພວກເຮົາໃຊ້ເຄື່ອງກວດອັດຕະໂນມັດ: ຜູ້ເຂົ້າຮ່ວມຈໍາເປັນຕ້ອງໄດ້ໄປຫາບັນຊີສ່ວນຕົວຂອງລາວ, ກົດປຸ່ມ "ກວດເບິ່ງ", ແລະຫຼັງຈາກເວລາໃດຫນຶ່ງລາວເຫັນບາງຄໍາຕິຊົມເພີ່ມເຕີມກ່ຽວກັບສິ່ງທີ່ລາວໄດ້ເຮັດ. ແລະມັນແມ່ນໃນເວລານີ້ທີ່ພວກເຮົາເລີ່ມເຂົ້າຫາບັນຫາຂອງພວກເຮົາ.

ການຢັ້ງຢືນຂອງຫ້ອງທົດລອງນີ້ແມ່ນມີໂຄງສ້າງເຊັ່ນນີ້: ພວກເຮົາສົ່ງຊຸດຂໍ້ມູນການຄວບຄຸມໄປຫາ Kafka ຂອງຜູ້ເຂົ້າຮ່ວມ, ຫຼັງຈາກນັ້ນ Gobblin ໂອນຊຸດຂໍ້ມູນນີ້ໄປຫາ HDFS, ຫຼັງຈາກນັ້ນ Airflow ເອົາຊຸດຂໍ້ມູນນີ້ແລະວາງໄວ້ໃນ ClickHouse. ເຄັດລັບແມ່ນວ່າ Airflow ບໍ່ຈໍາເປັນຕ້ອງເຮັດແບບນີ້ໃນເວລາຈິງ, ມັນເຮັດຕາມຕາຕະລາງ: ທຸກໆ 15 ນາທີມັນໃຊ້ເວລາຫຼາຍໄຟລ໌ແລະອັບໂຫລດພວກມັນ.

ມັນປະກົດວ່າພວກເຮົາຈໍາເປັນຕ້ອງໄດ້ກະຕຸ້ນ DAG ຂອງເຂົາເຈົ້າເອງຕາມຄໍາຮ້ອງຂໍຂອງພວກເຮົາໃນຂະນະທີ່ຕົວກວດສອບກໍາລັງແລ່ນຢູ່ທີ່ນີ້ແລະໃນປັດຈຸບັນ. ຫຼັງຈາກ googling, ພວກເຮົາພົບເຫັນວ່າສໍາລັບສະບັບຕໍ່ມາຂອງ 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) $

ຕິດຕັ້ງ Airflow:

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

ຂັ້ນຕອນຕໍ່ໄປແມ່ນເພື່ອດໍາເນີນການຄໍາສັ່ງທີ່ຈະສ້າງແລະເລີ່ມຕົ້ນຖານຂໍ້ມູນ dataflow ໃນ SQLite:

(venv) $ airflow initdb

ຖານຂໍ້ມູນຈະຖືກສ້າງຂື້ນໃນ airflow.db ຄ່າເລີ່ມຕົ້ນ.

ໃຫ້ກວດເບິ່ງວ່າ Airflow ຖືກຕິດຕັ້ງ:

$ 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 ທົດລອງ

ໃນຈຸດນີ້, Airflow ໄດ້ຖືກຕັ້ງຄ່າແລະພ້ອມທີ່ຈະໄປ. ຢ່າງໃດກໍຕາມ, ພວກເຮົາຍັງຈໍາເປັນຕ້ອງດໍາເນີນການ API ທົດລອງ. checkers ຂອງພວກເຮົາຖືກຂຽນໄວ້ໃນ Python, ດັ່ງນັ້ນການຮ້ອງຂໍທັງຫມົດຈະຢູ່ໃນມັນໂດຍໃຊ້ຫ້ອງສະຫມຸດ 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

ຫຼັງຈາກນັ້ນ, ທ່ານຈໍາເປັນຕ້ອງສ້າງຜູ້ໃຊ້ຂອງທ່ານດ້ວຍສິດທິ admin:

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