Airflow es una plataforma para la definición, ejecución y monitoreo de flujos de trabajo (Workflows) creada por AirBnB, liberada bajo licencia Apache, y que desde hace unos años es un proyecto en incubación de Apache Software Foundation.

Como parte de la evolución de CDCol, evaluamos alternativas para la definición de workflows dinámicos. Después de evaluar alternativas entre las que se encontraban Luigi, Pinball y un desarrollo propio basado en Celery, la interfaz gráfica, la integración con Celery -que hace parte de la arquitectura actual CDCol-, y características como XCom, inclinaron la balanza a favor de Airflow.

Airflow tiene, conceptualmente, cuatro componentes: Un servidor web, un scheduler, ejecutores (con sus workers) y la base de datos.

Este artículo describirá la instalación de Airflow con Celery y las primeras pruebas, incluyendo la definición de workflows (DAGs) y plugins.

Instalación

Entorno

  • Máquina virtual con Ubuntu Server 18.04
  • Anaconda 3.4

Instalación Standalone

La instalación de airflow por defecto, usando el ejecutor local y la base de datos sqlite (que no es recomendable en entornos de producción), usando Anaconda, se reduce a:

export AIRFLOW_HOME="$HOME/airflow"
echo 'export AIRFLOW_HOME="$HOME/airflow"' >>$HOME/.bashrc
conda install -y -c conda-forge airflow
airflow initdb

La primera vez que se ejecuta, initdb crea el archivo de configuración $AIRFLOW_HOME/airflow.cfg con los valores por defecto y una base de datos sqllite con los valores por omisión. En este punto se puede ejecutar airflow webserver -p 8080 para verlo en acción en http://<dirección ip de la máquina>:8080.

Es una instalación funcional, en una sola máquina, sin embargo no se acerca a lo que queremos en un entorno de producción. Lo primero que vamos a hacer es cambiar el motor de base de datos que se utilizará para almacenar los metadatos. Para esto vamos a crear una nueva base de datos en postgresql y un usuario que pueda acceder de manera remota.

Después de matar el servidor web de airflow, instalamos las nuevas dependencias y creamos tanto la base de datos como el usuario (se puede ser mucho más creativo en los nombres 😛 ):

sudo apt install -y  postgresql postgresql-client postgresql-contrib
conda install -y psycopg2

sudo -u postgres createdb airflow
sudo -u postgres createuser airflow
sudo -u postgres psql airflow -c "alter user airflow with encrypted password 'la contraseña'"
sudo -u postgres psql airflow -c "grant all privileges on database airflow to airflow;"

#Configurar postgresql para que admita conexiones remotas
_HBA=`sudo -u postgres psql -t -P format=unaligned -c 'show hba_file'`
_CONFIG=`sudo -u postgres psql -t -P format=unaligned -c 'show config_file'`
mkdir -p $HOME/pg_backup
sudo cp "$_HBA" $HOME/pg_backup
cp "$_CONFIG" $HOME/pg_backup

sudo su -c "echo 'host all all 0.0.0.0/0 md5' >>$_HBA"
sudo sed -i "s/#listen_addresses = 'localhost'/listen_addresses = '*'/" $_CONFIG

#Disclaimer: Escuchar desde todas las interfaces
#y permitir conexiones desde todas las direcciones
#no es una buena política

sudo systemctl restart postgresql.service

Una vez tenemos la base de datos creada, con los permisos adecuados, modificamos el archivo de configuración de airflow ($AIRFLOW_HOME/airflow.cfg) para que la utilice, modificando la propiedad sql_alchemy_conn. La cadena de conexión en este caso quedaría:

sql_alchemy_conn = postgresql+psycopg2://airflow:la contraseña@10.0.2.15:5432/airflow

En este mismo archivo, también es necesario cambiar el ejecutor, de SequentialExecutor a LocalExecutor


executor = LocalExecutor

Luego volvemos a ejecutar airflow initdb, para inicializar la base de datos. Si todo ha salido bien, la salida será similar a la imagen.

Debemos asegurarnos de que exista la carpeta dags dentro del home de airflow.

mkdir -p "$AIRFLOW_HOME/dags"

Por alguna razón el scheduler falla (al menos en la versión 1.9) si en la carpeta dags no hay por lo menos un DAG válido, así que, mientras creamos nuestros workflows, podemos crear un DAG dummy (dummy.py):

import airflow
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator

from datetime import timedelta

args = {
'owner': 'airflow',
'start_date': airflow.utils.dates.days_ago(2)
}

dag = DAG(
dag_id='example_dummy', default_args=args,
schedule_interval=None,
dagrun_timeout=timedelta(minutes=1))

run_this_last = DummyOperator(task_id='DOES_NOTHING', dag=dag)

Luego podemos iniciar el scheduler y el servidor web (la opción -D hace que se ejecuten como demonios):

airflow scheduler -D
airflow webserver -p 8080

Podemos, por ejemplo, ejecutar uno de los DAGs. Es importante fijarnos que no esté pausado (debe estar en On), de otra forma no se ejecutará.

Podemos ver el estado (si está corriendo, terminó correctamente o falló):

y el detalle de cada ejecución:

Esta instalación standalone es útil para desarrollo y un poco más cercana a lo que tendremos en producción que la instalación por defecto.
NOTA: Si esto es lo que necesitas, publiqué en GIST un script que genera este ambiente:

 

Instalación con Celery

Para hacer que airflow use como workers a los nodos de celery, basta con:

  1. Tener un cluster de celery funcional, usando como backend de resultados rabbitMQ o Redis. Airflow y celery son proyectos en desarrollo por lo que es importante verificar que las versiones sean compatibles entre los dos. En CDCol usamos celery version 3.x por lo que debemos usar una versión de airflow menor a 1.9 para poder tener más de un ejecutor sin inconvenientes.
  2. Modificar el archivo de configuración de airflow para que use el ejecutor `CeleryExecutor`.
  3. Instalar airflow en todos los nodos de celery. La configuración de airflow en todos los nodos debe ser homogénea.
  4. Asegurarse que las carpetas dags y plugins de airflow estén sincronizadas en todos los nodos.

Para este artículo, usaré Redis como backend de resultados y RabbitMQ como manejador de colas de mensajes.  Estarán instalados en la máquina en la que se ejecutará el scheduler y el webserver de airflow, sin embargo pueden estar en cualquier máquina que sea alcanzable por los nodos.

sudo apt install redis-server
sudo apt install rabbitmq-server

#Crear un usuario y un vhost para la cola de rabbit, habilitar el plugin para el monitoreo (requerido si se va a usar flower)

sudo rabbitmqctl add_user airflow airflow
sudo rabbitmqctl add_vhost airflow
sudo rabbitmqctl set_user_tags airflow airflow_tag administrator
sudo rabbitmqctl set_permissions -p airflow airflow ".*" ".*" ".*"
sudo rabbitmq-plugins enable rabbitmq_management

conda install redis-py
conda install -c conda-forge "celery<4" flower

Modificamos la configuración de celery para que use el ejecutor de celery, para que use redis server como backend de resultados y rabbitMQ como broker.

# The executor class that airflow should use. Choices include
# SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor
executor = CeleryExecutor


# The Celery broker URL. Celery supports RabbitMQ, Redis and experimentally
# a sqlalchemy database. Refer to the Celery documentation for more
# information.
broker_url = amqp://airflow:airflow@10.0.2.15/airflow

# Another key Celery setting
celery_result_backend = redis://10.0.2.15:6379/0

En todas las máquinas que tendrán workers de celery debe estar instalado airflow, celery y redis-py.

conda install redis-py
conda install -c conda-forge "airflow<1.9" "celery<4"

También debemos tener un mecanismo para sincronizar las carpetas dags y plugins de airflow, la opción más sencilla es usar un punto de montaje NFS (la ubicación de estos directorios se especifica en airflow.cfg). Es importante notar que el ambiente de python también debe mantenerse homogéneo.

Una vez tenemos todo configurado ejecutamos en la máquina maestra el webserver y el scheduler, y en los nodos que ejecutarán el trabajo airflow worker.


#En la máquina maestra
airflow webserver -p 8080 -D
airflow scheduler -D
#opcional:

airflow flower -D


#En cada nodo
airflow worker -D

Podemos resumir el proceso en dos scripts, uno para la máquina maestra (que en este caso también tiene un worker) y otro para los workers:

 

Para monitorear los workers de celery podemos usar flower, ejecutándolo con airflow flower -D y accediendo por el puerto 5555.

Definición de un Workflow

Una vez tenemos instalado airflow, lo siguiente que queremos hacer es definir un workflow. Airflow soporta únicamente grafos dirigidos acíclicos, lo que permite asegurar que, si todas las tareas terminan correctamente, el workflow termina (no hay ciclos infinitos). Los archivos de definición de DAGs en airflow son scripts de python.

El scheduler de airflow evalúa (ejecuta) periódicamente los archivos de la carpeta de DAGs, identificando cambios en la definición de los mismos. Es por esta razón que cada archivo de definición debe poder ejecutarse en segundos, de otro modo afectaría el rendimiento general de airflow.

En un archivo de definición de DAG se crea un objeto DAG y un conjunto de operadores que tienen dependencias entre ellos. El DAG debe tener un id, y puede definir una programación (cada cuánto debe ejecutarse), un timeout, una fecha de inicio, argumentos por defecto (que serán pasados a todas las tareas), y documentación. El DAG agrupa operadores.

Un operador describe una tarea. Se pueden agrupar en 3 categorías, según su objetivo: Operadores que ejecutan una acción, que transfieren datos, o sensores. Airflow tiene operadores que pueden cubrir la mayoría de necesidades, como BashOperator, PythonOperator, PostgresOperatorHTTPOperator`, entre otros.

Un ejemplo de workflow, que muestra el manejo de las dependencias, es:


import airflow
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator

from datetime import timedelta

def print_value(ds, value):
    print(value)
#Argumentos por defecto. 
args = {
'owner': 'airflow',
'start_date': airflow.utils.dates.days_ago(2)
}
#Definición del DAG
dag = DAG(
dag_id='test_print_values', default_args=args,
schedule_interval=None,
dagrun_timeout=timedelta(minutes=5))

#Tareas Dummy que pertenecen al DAG
start_with = DummyOperator(task_id='PRIMERA_TAREA', dag=dag)

end_with = DummyOperator(task_id='TAREA_FINAL', dag=dag)

#Crear 5 Tareas que se ejecutarán después de start_with y antes de end_with
#Se pueden usar los operadores >> y << para expresar las dependencias. for i in range(5): _po=PythonOperator(task_id="echo_{}".format(i), python_callable=print_value, op_kargs={'value':i}, dag=dag ) start_with>>_po>>end_with

Esta definición genera el workflow de la siguiente imagen:

Y al ejecutar el DAG se puede ver el progreso, donde se muestra que la tarea final sólo se ejecuta una vez todas las tareas anteriores han terminado:

 

Por defecto, los nuevos dags se crean pausados, no se ejecutarán hasta que se activen por la interfaz web o usando la interfaz de línea de comando. Un DAG que no tenga schedule_interval (o sea None), sólo se ejecutará al lanzarlo explícitamente desde la web o usando trigger_dag.


airflow trigger_dag test_print_values

Definición de un plugin

 

Aunque los operadores incluidos en Airflow son bastante genéricos, muchas veces necesitamos utilidades adicionales, nuevos tipos de operadores, o vistas web. Para esto, Airflow permite definir plugins. Un plugin se define con un módulo de python en la carpeta de plugins. Un plugin se define con una clase extiende airflow.plugins_manager.AirflowPlugin y define conjuntos de operadores, ejecutores, vistas, “hooks“, macros y links de menú.

Cada uno de esos elementos se define como una clase que extiende una base. Por ejemplo, un operador se define como una clase que extiende airflow.models.BaseOperator y un hook a airflow.hooks.base_hook.

Por ejemplo, podríamos definir un plugin con un operador que reciba una carpeta y genere un archivo tar.gz con la salida. Para eso definimos el operador, extendiendo de BaseOperator, con su constructor y el método execute, que es el que realiza el trabajo.

import tarfile
import os
from airflow.models import BaseOperator
from airflow import utils as airflow_utils

class TarGzOperator(BaseOperator):
    @airflow_utils.apply_defaults
    def __init__(self, to_compress, output_folder, *args,**kwargs):
        super().__init__(*args, **kwargs)
        if to_compress is None or output_folder is None: 
            raise Exception("You must specify the input and output folder")
        if to_compress == output_folder:
            raise Exception("to compress folder cannot be the output folder")
        if os.path.exists(output_folder) and not os.path.isdir(output_folder):
            raise Exception("output_folder must be a folder")
        self.to_compress = to_compress
        self.output_folder = output_folder
        
    
    def execute(self, context):
        print(context)
        _filename= os.path.basename(os.path.normpath(self.to_compress))
        _output_file=os.path.join(self.output_folder, _filename+str(".tar.gz") )
        if not os.path.exists(self.output_folder):
            os.makedirs(self.output_folder)
        with tarfile.open(_output_file,"w:gz" ) as _tar:
            _tar.add(self.to_compress)
            
        
        return _output_file

Luego es necesario definir el plugin :


from airflow.plugins_manager import AirflowPlugin

#si la definición del operador se hace en un archivo diferente,
# sería necesario hacer el import

class TarGzPlugin(AirflowPlugin):
    name = 'targz_plugin'
    operators=[TarGzOperator]

Al publicar el archivo en la carpeta plugins de airflow, y reiniciar tanto el webserver como el el scheduler, el nuevo plugin está disponible. Se pueden definir DAGs que lo usen:

import airflow
from airflow.models import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators import TarGzOperator

from datetime import timedelta

args = {
    'owner': 'airflow',
    'start_date': airflow.utils.dates.days_ago(2)
}

dag = DAG(
    dag_id='test_plugin', default_args=args,
    schedule_interval=None,
    dagrun_timeout=timedelta(minutes=5))

dummy = DummyOperator(task_id='DOES_NOTHING', dag=dag)

tarBackups = TarGzOperator(to_compress='/home/cronosnull/airflow',output_folder='/home/cronosnull',task_id='targz',dag=dag)

dummy >> tarBackups

Al habilitar y ejecutar el DAG desde la interfaz web se generará el archivo tar.gz esperado y se puede ver que el valor retornado por el operador queda registrado en XCom (lo que permitirá usarlo desde una siguiente tarea).

Airflow es una herramienta extensible, que se adapta a escenarios diversos. Espero que este artículo sea útil para quien está empezando a explorarla, le ahorre algo de tiempo y le despierte nuevas dudas -mientras lo escribía resolví algunas-.

En un post anterior realizamos la instalación de hadoop en un nodo, tanto en modo standalone como pseudo-distribuido. Ahora realizaremos la instalación en dos nodos.

Hadoop Logo

Un par de cosas a tener en cuenta:

  • De acuerdo a la documentación, típicamente se tiene un nodo como NameNode y otro como JobTracker. Los demás nodos serán configurados como DataNode y TaskTracker.
  • Aunque no es obligatorio, es recomendable que los nodos tengan el mismo HADOOP_HOME, directorio raíz de la distribución.

Lo primero por hacer es descargar Hadoop, descomprimirlo en lo que será nuestro HADOOP_HOME, y configurar el JAVA_HOME dentro de conf/hadoop-env.sh.

wget http://apache.mirrors.tds.net//hadoop/core/stable/hadoop-0.20.2.tar.gz
tar -xvf hadoop-0.20.2.tar.gz
vim hadoop-0.20.2/conf/hadoop-env.sh

Para distribuir un poco la carga configuraremos el NameNode en una máquina y el JobTracker en otra. También configuraremos los dos nodos como TaskTracker y DataNode.

Para ello debemos modificar el archivo de configuración conf/core-site.xml con la url del que será el NameNode.

<configuration>

<property>

<name>fs.default.name</name>

<value>hdfs://namenode:9000</value>

</property>

</configuration>

Para especificarle a Hadoop en donde almacenar los bloques de datos del DFS se puede usar la propiedad dfs.data.dir y como valor la ruta del directorio, en el archivo conf/hdfs-site.xml. En éste archivo es necesario configurar la replicación de los bloques, el valor predeterminado es 3, en éste caso solo tenemos dos nodos así que utilizaremos un valor de dfs.replication de 2.

<configuration>
<property>
<name>dfs.data.dir</name>
<value>/data/hadoopdd</value>
</property>
<property>

<name>dfs.name.dir</name>

<value>/data/hadoopnd</value>

</property>
<property>
<name>dfs.replication</name>
<value>2</value>
</property>
</configuration>

Debemos especificar la ubicación del JobTracker, lo hacemos en el archivo de configuración mapred-site.xml. En este archivo también configuramos donde se almacenarán los archivos temporales.

<configuration>

<property>

<name>mapred.job.tracker</name>

<value>157.253.236.228:54311</value>

</property>

<property>

<name>mapred.local.dir</name>

<value>/data/hadoopmr</value>

</property>

</configuration>

Debemos asegurarnos que podemos iniciar sesión ssh sin contraseña en las dos máquinas, para ello seguimos los pasos indicados para la instalación en un solo nodo y, además, agregar las llaves públicas ssh de las máquinas master en el archivo .ssh/authorized_keys máquinas slave:

$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa

$ cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

$ ssh-copy-id -i .ssh/id_dsa.pub {ip_slave}

Los siguientes archivos solo se deben editar en los master(en este caso en los dos nodos, pero al agregar un nuevo nodo solo se modifican en los NameNode y JobTracker).

En el archivo conf/masters se listan los nodos master, tanto NodeName como JobTracker, en el archivo conf/slaves todos los nodos DataNode y TaskTracker, uno por línea.

Formateamos el sistema de archivos distribuido, ejecutando en el NameNode:

./hadoop-0.20.2/bin/hadoop namenode -format

Arrancamos los demonios de hdfs, ejecutando

bin/start-dfs.sh

Podemos los procesos ejecutados en cáda maquina usando el comando jps.

A continuación iniciamos los demonios de Mapreduce, ejecutando en el JobTracker:

bin/start-mapred.sh

Para detener los demonios del sistema de archivos y de mapreduce, respectivamente, ejecutamos:

bin/stop-dfs.sh
bin/stop-mapred.sh

Ahora deberíamos poder correr el ejemplo que utilizamos en la instalación anterior:

$ bin/hadoop fs -put conf input

$ bin/hadoop jar hadoop-*-examples.jar grep input/conf output 'dfs[a-z.]+'

Sin embargo se pueden presentar algunos problemas en el dfs, por un bug que aún se encuentra abierto. En http://www.michael-noll.com/wiki/Running_Hadoop_On_Ubuntu_Linux_(Multi-Node_Cluster) podemos ver otro tutorial(en ingles) de la instalación del cluster, con una configuración un poco diferente, y las alternativas para resolver los problemas que se pueden presentar.

En un próximo artículo revisaremos el desarrollo de una aplicación MapReduce para Hadoop.

 

Hadoop Logo

Hadoop es un framework para computación distribuida que soporta aplicaciones con uso intensivo de datos. Implementa, entre otras cosas, el paradigma MapReduce y HDFS, un sistema de archivos distribuido y el principal sistema de almacenamiento en Hadoop.

En esta primera aproximación probaré la instalación y ejecución de Hadoop, haciendo un resumen de los pasos necesarios, traduciéndolos, y comentarios al margen frente a la documentación.

Los prerrequisitos de Hadoop son Java 1.6, ssh server y rsync, en Windows será necesario también Cygwin. Para éstas pruebas usaré una máquina virtual con Debian 5.0.5, Java 1.6.0_20, openssh 5.1p1.

La instalación de Hadoop se reduce a descargar la versión estable y descomprimirlo:

wget http://apache.mirrors.tds.net//hadoop/core/stable/hadoop-0.20.2.tar.gz
tar -xvf hadoop-0.20.2.tar.gz
Ahora, entrando a la carpeta que se descomprimió, se edita conf/hadoop-env.sh definiendo la variable JAVA_HOME con la ubicación de la instalación de java, en teoría funcionaría si la variable JAVA_HOME está definida en el entorno, pero es mejor definirla en el archivo para facilitar la distribución. A continuación se puede ejecutar el comando:
bin/hadoop
Para ver  la documentación de uso.
Existen 3 configuraciones posibles de Hadoop: Standalone, pseudo-distribuida y distribuida. La configuración predeterminada es Standalone, no distribuida y un solo proceso de java, útil para depuración según la documentación. Con ésta instalación se puede ejecutar el primer ejemplo de Hadoop, un grep basado en MapReduce:
$ mkdir input
$ cp conf/*.xml input
$ bin/hadoop jar hadoop-*-examples.jar grep input output 'dfs[a-z.]+'
$ cat output/*

La salida crea dos archivos: part-0000 con la respuesta y .part-0000.crc con el CRC checksum.

Para ejecutar Hadoop en el modo Pseudo-distribuido, varios nodos en una máquina en procesos java diferentes, debemos modificar los archivos conf/core-site.xml, conf/hdfs-site.xml y conf/mapred-site.xml así:

conf/core-site.xml:

<configuration>

<property>

<name>fs.default.name</name>

<value>hdfs://localhost:9000</value>

</property>

</configuration>

conf/hdfs-site.xml:

<configuration>

<property>

<name>dfs.replication</name>

<value>1</value>

</property>

</configuration>

conf/mapred-site.xml:

<configuration>

<property>

<name>mapred.job.tracker</name>

<value>localhost:9001</value>

</property>

</configuration>

En el ambiente de pruebas funciona ssh a localhost sin contraseña, sin embargo esto puede no ser cierto en todos los ambientes. En el manual de Hadoop describen la forma configurarlo para ese caso. Esto sólo es necesario si no se puede hacer ssh a localhost sin contraseña:

$ ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa

$ cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

Ahora podemos ejecutar hadoop, primero creando un nuevo sistema de archivo distribuido, para luego ejecutar los nodos

$ bin/hadoop namenode -format
$ bin/start-all.sh

Ahora podemos acceder a la interfaz web del Job Tracker y del Namenode en el puerto 50030 y 50070 respectivamente.

Ahora ejecutaremos el ejemplo con el que probamos el Singlenode, para lo cual necesitamos copiar los archivos de entrada al sistema de archivos distribuido, en el cual también se crearán los archivos de salida, para hacer visibles las diferencias entre los dos modos se pueden borrar las carpetas input y output creadas anteriormente. Para ejecutar el ejemplo hacemos:

$ rm -rf input output

$ bin/hadoop fs -put conf input

$ bin/hadoop jar hadoop-*-examples.jar grep input output 'dfs[a-z.]+'

Ahora los archivos de salida están en el sistema de archivos distribuido, para traerlos ejecutamos:

$ bin/hadoop fs -get output output
$ cat output/*

o podemos ver el resultado directamente en el sistema de archivos distribuido:

bin/hadoop fs -cat output/*

Cuando terminamos podemos detener los nodos ejecutando:

$ bin/stop-all.sh

La salida incluye además de la respuesta los logs de la ejecución.

En un próximo post revisaremos la configuración de Hadoop en modo distribuido en un Cluster de máquinas.

Hadoop Logo