yum -y install mysql-devel gcc gcc-devel python-devel gcc-c++ cyrus-sasl cyrus-sasl-devel cyrus-sasl-lib 

vim /etc/profile
export AIRFLOW_HOME=/root/airflow

#Сделаем переменные среды Конфигурации эффективными
source /etc/profile

(python37)   conda activate python37
(python37) pip install apache-airflow==2.1.3 -i https://pypi.tuna.tsinghua.edu.cn/simple

(python37)  airflow version
2.1.3

CREATE DATABASE airflow CHARACTER SET utf8;
create user 'airflow'@'%' identified by '123456';
grant all privileges on airflow.* to 'airflow'@'%';
flush privileges;

[mysqld]
explicit_defaults_for_timestamp=1

#Перезапускаем MySQL
[root@node2 ~]# service mysqld restart

#Повторно войдите в запрос MySQL
mysql> show variables like 'explicit_defaults_for_timestamp';

[core]
dags_folder = /root/airflow/dags

#Изменить часовой пояс
default_timezone = Asia/Shanghai

#Конфигурация Тип Executor, рекомендация по кластеру Конфигурация CeleryExecutor
executor = CeleryExecutor

# Конфигурациябаза данных
sql_alchemy_conn=mysql+mysqldb://airflow:123456@node2:3306/airflow?use_unicode=true&charset=utf8

[webserver]
#Установить часовой пояс
default_ui_timezone = Asia/Shanghai

[celery]
#КонфигурацияCelery Очередь сообщений, используемая брокером
broker_url = redis://node4:6379/0
#КонфигурацияCelery После завершения задачи брокера статус обновляется с помощью библиотеки
result_backend = db+mysql://root:123456@node2:3306/airflow

(python37) [root@node1 airflow]# scp ./airflow.cfg  node2:`pwd`
(python37) [root@node1 airflow]# scp ./airflow.cfg  node3:`pwd`
(python37) [root@node1 airflow]# scp ./airflow.cfg  node4:`pwd`

(python37) #  pip install mysqlclient -i https://pypi.tuna.tsinghua.edu.cn/simple

(python37) [root@node1 airflow]# airflow db init

airflow users create \
    --username airflow \
    --firstname airflow \
    --lastname airflow \
    --role Admin \
    --email xx@qq.com

(python37) [root@node1 software]# yum -y install unzip
(python37) [root@node1 software]# unzip ./airflow-scheduler-failover-controller-master.zip

(python37) [root@node1 software]# cd /software/airflow-scheduler-failover-controller-master
(python37) [root@node1 airflow-scheduler-failover-controller-master]# pip install -e .

(python37) [root@node1 ~]# scheduler_failover_controller init
Adding Scheduler Failover configs to Airflow config file...
Finished adding Scheduler Failover configs to Airflow config file.
Finished Initializing Configurations to allow Scheduler Failover Controller to run. Please update the airflow.cfg with your desired configurations.

[scheduler_failover]
# Конфигурацияairflow Главный узел, где Конфигурация — node1, node2, оба узла не должны иметь паролей.
scheduler_nodes_in_cluster = node1,node2

#В строке 1088 обратите особое внимание на то, что точку с запятой необходимо убрать, иначе Планировщик не сможет нормально запуститься при его автоматическом перезапуске позже.
airflow_scheduler_start_command = export AIRFLOW_HOME=/root/airflow;nohup airflow scheduler >> ~/airflow/logs/scheduler.logs &

(python37) [root@node1 airflow]# scheduler_failover_controller test_connection
Testing Connection for host 'node1'
(True, ['Connection Succeeded', ''])
Testing Connection for host 'node2'
(True, ['Connection Succeeded\n'])

(python37) [root@node1 ~]# cd /root/airflow/
(python37) [root@node1 airflow]# scp airflow.cfg node2:`pwd`
(python37) [root@node1 airflow]# scp airflow.cfg node3:`pwd`
(python37) [root@node1 airflow]# scp airflow.cfg node4:`pwd`

(python37) [root@node1 airflow]# pip install celery==4.4.7 flower==0.9.7 redis==3.5.3
(python37) [root@node2 airflow]# pip install celery==4.4.7 flower==0.9.7 redis==3.5.3
(python37) [root@node3 airflow]# pip install celery==4.4.7 flower==0.9.7 redis==3.5.3
(python37) [root@node4 airflow]# pip install celery==4.4.7 flower==0.9.7 redis==3.5.3

#Фоновый запуск по умолчанию может использовать -D , используя здесь -D, иногда невозможно нормально запустить соответствующий процесс Airflow.
airflow webserver
airflow scheduler

airflow webserver

(python37) [root@node3 ~]# airflow celery worker
(python37) [root@node4 ~]# airflow celery worker

(python37) [root@node1 airflow]# nohup scheduler_failover_controller start > /root/airflow/logs/scheduler_failover/scheduler_failover_run.log &

#!/bin/bash

dt=$1

echo "==== execute first shell ===="

echo "---- first : time is ${dt}"

#!/bin/bash

dt=$1

echo "==== execute second shell ===="

echo "---- second : time is ${dt}"

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {
    'owner':'zhangsan',
    'start_date':datetime(2021, 9, 23),
    'retries': 1,  # Количество неудачных попыток
    'retry_delay': timedelta(minutes=5) # Интервал повторения ошибки
}

dag = DAG(
    dag_id = 'execute_shell_sh',
    default_args=default_args,
    schedule_interval=timedelta(minutes=1)
)

first=BashOperator(
    task_id='first',
    #Рекомендуется прописывать абсолютный путь к скрипту
    bash_command='sh /root/airflow/dags/first_shell.sh %s'%datetime.now().strftime("%Y-%m-%d"),
    dag = dag
)

second=BashOperator(
    task_id='second',
    #Рекомендуется прописывать абсолютный путь к скрипту
    bash_command='sh /root/airflow/dags/second_shell.sh %s'%datetime.now().strftime("%Y-%m-%d"),
    dag=dag
)

first >> second

(python37) [root@node1 dags]# ps aux |grep webserver
(python37) [root@node1 dags]# ps aux |grep scheduler
(python37) [root@node2 dags]# ps aux |grep webserver
(python37) [root@node2 dags]# ps aux |grep scheduler
(python37) [root@node3 ~]# ps aux|grep "celery worker"
(python37) [root@node4 ~]# ps aux|grep "celery worker"
Найдите номер процесса, соответствующий соответствующей команде запуска, и завершите его.

(python37) [root@node1 ~]# ps aux|grep scheduler

root      23744  0.9  3.3 326940 63028 pts/2    S    00:08   0:02 airflow scheduler -- DagFileProcessorManager

#kill Отменить процесс планировщика

(python37) [root@node1 ~]# kill -9 23744



#Доступ к веб-серверу webui