When preparing our educational programs, we occasionally encounter difficulties regarding some tools. And at the moment we face these challenges, there is not always enough documentation and articles to help us tackle these problems.
This was the case, for example, in 2015 when we used a Hadoop cluster with Spark for our "Big Data Specialist" program, accommodating 35 simultaneous users. It was unclear how to prepare it for such a use case using YARN. Ultimately, after figuring it out and navigating the process ourselves, we created and also presented at .
Background
This time, we will talk about another program – . In this program, our participants build two types of architectures: lambda and kappa. In the lambda architecture, for batch processing, Airflow is used to transfer logs from HDFS to ClickHouse.
Overall, everything is going well. Let them build their pipelines. However, there is a catch: all our programs are technologically advanced in terms of the learning process itself. To check labs, we use automated checkers: a participant needs to log into their personal account, click the "Check" button, and after a while, they receive some detailed feedback on what they did. It is precisely at this moment that we begin to approach our problem.
The verification of this lab is organized as follows: we send a control data packet to the participant's Kafka, then Gobblin transfers this data packet to HDFS, and afterwards, Airflow takes this data packet and places it into ClickHouse. The trick is that Airflow does not need to do this in real-time; it operates on a schedule: every 15 minutes it takes a batch of files and uploads them.
This means we need to somehow trigger their DAG ourselves on demand during the checker’s operation here and now. After some googling, we found that for later versions of Airflow, there is something called . The word experimental, of course, sounds frightening, but what can we do... Perhaps it will take off.
Next, we will describe the entire journey: from installing Airflow to forming a POST request that triggers the DAG using the Experimental API. We will be working with Ubuntu 16.04.
1. Installing Airflow
We will check that we have Python 3 and virtualenv installed.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0If any of this is missing, please install it.
Now let's create a directory where we will work with Airflow.
$ mkdir
$ cd /path/to/your/new/directory
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Let's install Airflow:
(venv) $ pip install airflowThe version we were working with: 1.10.
Now we need to create a directory airflow_home, where the DAG files and Airflow plugins will be located. After creating the directory, we will set the environment variable AIRFLOW_HOME.
(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=The next step is to execute the command that will create and initialize the data flow database in SQLite:
(venv) $ airflow initdbThe database will be created in airflow.db by default.
Let's check if Airflow is installed:
$ 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.0If the command executed successfully, then Airflow has created its configuration file airflow.cfg downward API support (simultaneously with this in AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAirflow has a web interface. It can be started with the command:
(venv) $ airflow webserver --port 8081Now you can access the web interface in your browser on port 8081 on the host where Airflow was started, for example: <hostname:8081>.
2. Working with the Experimental API
At this point, Airflow is configured and ready for use. However, we also need to start the Experimental API. Our checkers are written in Python, so all subsequent requests will be made in it using the library requests.
In fact, the API is already working for simple requests. For example, this request can be used to test its functionality:
>>> import requests
>>> host =
>>> airflow_port = 8081 # in our case, this one, but by default 8080
>>> requests.get('http://{}:{}//{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'If you received this message in response, it means that everything is working.
However, when we want to trigger a DAG, we will encounter the issue that this type of request cannot be made without authentication.
For that, we will need to perform a few additional steps.
First, we need to add this to the config:
[api]
auth_backend = airflow.contrib.auth.backends.password_authThen, we need to create our user with admin rights:
>>> 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()Next, you need to create a user with regular permissions who will be allowed to trigger the 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()Now everything is ready.
3. Initiating the POST request
The POST request will look like this:
>>> 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 "n}n'The request has been processed successfully.
Accordingly, we then give the DAG some time to process and make a request to the ClickHouse table, trying to catch the control data packet.
The check is complete.
Source: habr.com
