Cómo configurar un clúster de flujo de aire multinodo con Celery y RabbitMQ

¿Qué es Airflow?

Apache Airflow es un administrador de flujo de trabajo avanzado y una herramienta indispensable en el arsenal del ingeniero de datos moderno.

Airflow le permite crear flujos de trabajo en forma de gráficos acíclicos dirigidos (DAG) de tareas. Una variedad de utilidades de línea de comando realizan operaciones complejas en el DAG. La interfaz de usuario visualiza fácilmente las canalizaciones que se ejecutan en un entorno de producción, monitorea el progreso y soluciona problemas según sea necesario.

Cree, planifique y controle su flujo de trabajo mediante programación. Proporciona una abstracción funcional en forma de un DAG idempotente (gráfico acíclico dirigido). Una función como un servicio de abstracción para realizar tareas a intervalos específicos.

Clúster con un nodo Airflow

En un clúster de Airflow de un solo nodo, todos los componentes (trabajador, programador, servidor web) se instalan en un solo nodo conocido como " nodo principal ". Para escalar un clúster de un solo nodo se Airflow



debe configurar en LocalExecutor



. El trabajador toma (extrae) una tarea de la cola de IPC (comunicación entre procesos), esto escala muy bien siempre que los recursos estén disponibles en el nodo maestro . Para escalar Airflow en varios nodos, debe habilitar Celery Executor



.

Arquitectura de nodo único de flujo de aire

Clúster multinodo de flujo de aire

Airflow . - , , , . , Airflow CeleryExecutor



.

Celery CeleryExecutor



Airflow. / Celery Redis RabbitMQ. RabbitMQ — . — . IPC, , RabbitMQ — . RabbitMQ / , Celery . .

Airflow

Celery:

Celery — , . , . Airflow . Airflow Airflow, .

Airflow Celery:

. CentOS 7 Linux.

1. RabbitMQ

yum install epel-release
yum install rabbitmq-server
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

1. RabbitMQ Server

systemctl enable rabbitmq-server.service
systemctl start rabbitmq-server.service
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

1. - RabbitMQ

rabbitmq-plugins enable rabbitmq_management
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

rabbitmq — 15672



, - — admin/admin



.

1. pyamqp
   
   
   
   RabbitMQ PostGreSQL

pip install pyamqp
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

amqp://



— , librabbitmq, , py-amqp



, .

pyamqp://



librabbitmq://



, , . pyamqp://



amqp



(http://github.com/celery/py-amqp)

PostGreSQL: psycopg2

Psycopg — PostgreSQL Python.

pip install psycopg2
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

1. Airflow.

pip install 'apache-airflow[all]'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

airflow

airflow version
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Airflow v1.10.0, .

airflow initdb
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . Airflow .

1. Celery

Celery .

pip install celery==4.3.0
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Celery

celery --version
4.3.0 (rhubarb)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

1. airflow.cfg Celery Executor.

executor = CeleryExecutor
sql_alchemy_conn = postgresql+psycopg2://airflow:airflow@{HOSTNAME}/airflow 
broker_url= pyamqp://guest:guest@{RabbitMQ-HOSTNAME}:5672/
celery_result_backend = db+postgresql://airflow:airflow@{HOSTNAME}/airflow 
dags_are_paused_at_creation = True
load_examples = False
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

airflow.cfg



airflow airflow initdb



, airflow



.

- airflow

# default port is 8080
airflow webserver -p 8000
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

# start the scheduler
airflow scheduler
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

airflow .

airflow worker
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Una vez que haya terminado de iniciar los diversos servicios de flujo de aire, puede consultar la fantástica interfaz de flujo de aire con el comando:

http://<IP-ADDRESS/HOSTNAME>:8000
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

como especificamos el puerto 8000 en nuestro comando de inicio del servicio del servidor web, de lo contrario, el número de puerto predeterminado es 8080.

¡Si! Hemos terminado de crear un clúster con la arquitectura multinodo Airflow. :)