Experimental API를 사용하여 Airflow에서 DAG 트리거를 만드는 방법

교육 프로그램을 준비할 때 일부 도구 작업과 관련하여 주기적으로 어려움을 겪습니다. 그리고 우리가 그것들을 만나는 순간, 이 문제에 대처하는 데 도움이 될 충분한 문서와 기사가 항상 있는 것은 아닙니다.

예를 들어 2015년에는 Big Data Specialist 프로그램에서 35명의 동시 사용자를 위해 Spark와 함께 Hadoop 클러스터를 사용했습니다. YARN을 사용하여 이러한 사용자 사례를 준비하는 방법이 명확하지 않았습니다. 그 결과 그들은 스스로 길을 찾아 걸어갔고 하브레에 게시 또한 수행 모스크바 스파크 밋업.

선사 시대

이번에는 다른 프로그램에 대해 이야기하겠습니다. 데이터 엔지니어. 그 위에 참가자들은 람다와 카파라는 두 가지 유형의 아키텍처를 구축합니다. 그리고 lamdba 아키텍처에서 Airflow는 일괄 처리의 일부로 사용되어 HDFS에서 ClickHouse로 로그를 전송합니다.

모든 것이 일반적으로 좋습니다. 그들이 파이프라인을 구축하게 하십시오. 그러나 "그러나"가 있습니다. 우리의 모든 프로그램은 학습 과정 자체에서 기술적으로 진보합니다. 실험실을 확인하기 위해 자동 검사기를 사용합니다. 참가자는 개인 계정으로 이동하여 "확인" 버튼을 클릭하고 잠시 후 자신이 수행한 작업에 대한 확장된 피드백을 확인해야 합니다. 그리고 이 시점에서 우리는 문제에 접근하기 시작합니다.

이 랩을 확인하는 것은 다음과 같이 구성됩니다. 제어 데이터 패킷을 참가자의 Kafka로 보낸 다음 Gobblin이 이 데이터 패킷을 HDFS로 전송한 다음 Airflow가 이 데이터 패킷을 가져와 ClickHouse에 넣습니다. 요령은 Airflow가 이 작업을 실시간으로 수행할 필요가 없고 일정에 따라 수행한다는 것입니다. 15분에 한 번씩 많은 파일을 가져와 업로드합니다.

체커가 지금 여기에서 실행되는 동안 우리의 요청에 따라 어떻게든 DAG를 자체적으로 트리거해야 한다는 것이 밝혀졌습니다. 인터넷 검색을 통해 최신 버전의 Airflow에는 소위 실험적 API. 워드 experimental, 물론 무섭게 들리지만 어떡해 ... 갑자기 이륙합니다.

다음으로 Airflow 설치부터 Experimental API를 사용하여 DAG를 트리거하는 POST 요청 생성까지의 전체 경로를 설명합니다. 우리는 우분투 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가 설치되어 있는지 확인하십시오.

$ 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

이제 다음과 같이 Airflow가 실행 중인 호스트의 포트 8081에 있는 브라우저에서 웹 인터페이스에 액세스할 수 있습니다. <hostname:8081>.

2. 실험적 API 작업

이 Airflow에서 구성되어 사용할 준비가 되었습니다. 그러나 Experimental API도 실행해야 합니다. 우리 체커는 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

그런 다음 관리자 권한이 있는 사용자를 생성해야 합니다.

>>> 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 테이블을 처리하고 요청하여 제어 데이터 패킷을 포착하려고 시도할 시간을 줍니다.

확인이 완료되었습니다.

출처 : habr.com