Al preparar nuestros programas educativos, nos encontramos periódicamente con dificultades en la utilización de ciertas herramientas. Y en el momento en que nos encontramos con ellas, no siempre hay suficiente documentación y artículos que ayuden a resolver el problema.
Por ejemplo, esto sucedió en 2015, cuando en el programa 'Especialista en grandes datos' utilizamos un clúster Hadoop con Spark para 35 usuarios simultáneos. No estaba claro cómo prepararlo para este caso de uso utilizando YARN. Al final, después de resolverlo y recorrer el camino por nuestra cuenta, hicimos y también participamos en .
Antecedentes
Esta vez hablaremos de otro programa – . En él, nuestros participantes construyen dos tipos de arquitecturas: lambda y kappa. En la arquitectura lambda, durante el procesamiento por lotes, se utiliza Airflow para transferir registros de HDFS a ClickHouse.
Todo está bien en general. Que construyan sus pipelines. Sin embargo, hay un pero: todos nuestros programas son tecnológicos en términos del propio proceso de aprendizaje. Para verificar las prácticas, utilizamos verificadores automáticos: el participante necesita acceder a su cuenta personal, presionar el botón 'Verificar', y después de un tiempo recibe algún tipo de retroalimentación ampliada sobre lo que hizo. Y es en ese momento cuando comenzamos a acercarnos a nuestro problema.
La verificación de esta práctica está diseñada de la siguiente manera: enviamos un paquete de datos de control a Kafka del participante, luego Gobblin transfiere este paquete de datos a HDFS, después Airflow toma este paquete de datos y lo coloca en ClickHouse. La idea es que Airflow no debe hacerlo en tiempo real; lo hace según un horario: cada 15 minutos toma un lote de archivos y los sube.
Resulta que necesitamos de alguna manera activar su DAG de forma independiente bajo nuestra demanda durante el trabajo del verificador aquí y ahora. Al buscar en Google, descubrimos que para versiones más recientes de Airflow existe lo que se llama . La palabra experimental, por supuesto, suena aterradora, pero qué se le va a hacer… Tal vez funcione.
A continuación, describiremos todo el proceso: desde la instalación de Airflow hasta la formación de una solicitud POST que activa el DAG, utilizando la Experimental API. Trabajaremos con Ubuntu 16.04.
1. Instalación de Airflow
Comprobaremos que tenemos Python 3 y virtualenv instalados.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Si falta algo de esto, instálelo.
Ahora crearemos un directorio en el que trabajaremos con Airflow a partir de ahora.
$ mkdir
$ cd /ruta/a/tu/nueva/carpeta
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Instalemos Airflow:
(venv) $ pip install airflowLa versión que usamos fue: 1.10.
Ahora necesitamos crear un directorio airflow_home, donde se ubicarán los archivos DAG y plugins de Airflow. Después de crear el directorio, establezcamos la variable de entorno AIRFLOW_HOME.
(venv) $ cd /ruta/a/mi/espacio/airflow
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=El siguiente paso es ejecutar el comando que creará e inicializará la base de datos del flujo de datos en SQLite:
(venv) $ airflow initdbLa base de datos se creará en airflow.db por defecto.
Verifiquemos si Airflow se instaló correctamente:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Usando el ejecutor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generando tablas de gramática desde /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generando tablas de gramática desde /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / /
___ ___ | / _ / _ __/ _ / / _/ _| |/|/ /
_/_/_ |_/_/_ /_/ /_/_/ /_/_/ ____/____|__/
v1.10.0Si el comando se ejecutó correctamente, Airflow creó su archivo de configuración airflow.cfg en AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAirflow tiene una interfaz web. Se puede iniciar con el siguiente comando:
(venv) $ airflow webserver --port 8081Ahora puedes acceder a la interfaz web en el navegador en el puerto 8081 en el host donde se inició Airflow, por ejemplo: <hostname:8081>.
2. Trabajando con la API Experimental
En este punto, Airflow está configurado y listo para usar. Sin embargo, también necesitamos iniciar la API Experimental. Nuestros chequeadores están escritos en Python, así que a partir de ahora todas las solicitudes se realizarán en él utilizando la biblioteca requests.
De hecho, la API ya está funcionando para solicitudes simples. Por ejemplo, esta solicitud permite probar su funcionamiento:
>>> import requests
>>> host =
>>> airflow_port = 8081 #en nuestro caso es así, pero por defecto es 8080
>>> requests.get('http://{}:{}/*{}'.format(host, airflow_port, 'api/experimental/test').text
'OK'Si recibiste este mensaje en respuesta, significa que todo está funcionando.
Sin embargo, cuando queramos activar un DAG, nos encontraremos con que este tipo de solicitud no se puede hacer sin autenticación.
Para ello, será necesario realizar una serie de pasos adicionales.
Primero, hay que agregar esto a la configuración:
[api]
auth_backend = airflow.contrib.auth.backends.password_authLuego, necesitamos crear un usuario propio con derechos de administrador:
>>> 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()Luego, es necesario crear un usuario con derechos comunes al que se le permita activar el 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()Ahora todo está listo.
3. Ejecutar la solicitud POST
La solicitud POST se verá así:
>>> 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'La solicitud se procesó correctamente.
Por lo tanto, a continuación, le damos algo de tiempo al DAG para procesar y hacemos una consulta en la tabla ClickHouse, intentando capturar un paquete de datos de control.
La verificación ha finalizado.
Fuente: habr.com
