知识共享许可协议
本作品采用知识共享署名-非商业性使用-禁止演绎 3.0 未本地化版本许可协议进行许可。

更新源
yum install https://centos6.iuscommunity.org/ius-release.rpm -y
wget -O /etc/yum.repos.d/epel.repo http://mirrors.aliyun.com/repo/epel-6.repo
yum makecache
安装 Python3.6
yum install python36u python36u-devel -y
ln -s /usr/bin/python3.6 /bin/python3
安装 pip
yum install python36u-pip -y
ln -s /usr/bin/pip3.6 /bin/pip3
pip3 install --upgrade pip
安装 virtualenv
pip3 install virtualenv
安装 Airflow
  • 创建目录

    mkdir -p /opt/airflow
    mkdir -p /opt/airflow/airflow
    
    cd /opt/airflow/
    
  • 激活环境

    virtualenv -p `which python3.6` venv
    source venv/bin/activate
    
    退出虚拟环境
    deactivate
    
  • 安装依赖包

    pip3 install cryptography
    pip3 install flask-bcrypt
    pip3 install mysql-connector
    
  • 安装

    pip3 config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
    
    export AIRFLOW_HOME=/opt/airflow/airflow
    export SLUGIFY_USES_TEXT_UNIDECODE=yes
    pip3 install apache-airflow==1.10.2
    
    AIRFLOW_HOME=/opt/airflow/airflow SLUGIFY_USES_TEXT_UNIDECODE=yes pip3 install apache-airflow==1.10.2
    
配置
  • 修改时区

    vim $AIRFLOW_HOME/airflow.cfg
    
    default_timezone = Asia/Shanghai
    expose_config = True
    
    python -c "from airflow.utils import timezone;print(timezone.datetime(2019,7,1))"
    
  • 配置MySQL数据库

    • 安装包

      sudo apt-get install libmysqlclient-dev
      pip3 install apache-airflow[mysql]
      
    • 创建数据库

      CREATE DATABASE airflow;
      ALTER DATABASE `airflow` CHARACTER SET utf8; 
      CREATE USER 'airflow'@'%' IDENTIFIED BY 'af123854';
      GRANT all privileges on airflow.* TO 'airflow'@'%' IDENTIFIED BY 'af123854';
      FLUSH PRIVILEGES;
      
    • 修改cfg

      vim $AIRFLOW_HOME/airflow.cfg
      
      sql_alchemy_conn = mysql://airflow:af123854@192.168.1.100/airflow
      
  • 配置 celery + rabbitmq

    • 安装包

      pip3 install apache-airflow[celery] 
      pip3 install apache-airflow[rabbitmq]
      
    • 创建用户相关信息

      sudo rabbitmqctl add_user airflow af123854
      sudo rabbitmqctl add_vhost airflow-rabbitmq
      sudo rabbitmqctl set_user_tags airflow airflow-rabbitmq
      sudo rabbitmqctl set_permissions -p monitoring airflow ".*" ".*" ".*"
      
    • 修改cfg

      vim $AIRFLOW_HOME/airflow.cfg
      
      executor = CeleryExecutor
      
      broker_url = amqp://airflow:af123854@10.10.121.71:5672/airflow-rabbitmq
      airflow 1.9.0以上使用的是celery4.x, 而celery 4.x使用json序列化,不是用pickle进行序列化,修改为:
      broker_url = pyamqp://airflow:af123854@10.10.121.71:5672/airflow-rabbitmq
      
      result_backend 推荐使用数据库,在MySQL那边创建
      CREATE DATABASE result_backend;
      ALTER DATABASE `result_backend` CHARACTER SET utf8;
      GRANT all privileges on result_backend.* TO 'airflow'@'%' IDENTIFIED BY 'af123854';
      FLUSH PRIVILEGES;
      
      vim $AIRFLOW_HOME/airflow.cfg
      result_backend = db+mysql://airflow:af123854@10.10.132.132/result_backend
      
  • commands

    • init

      airflow initdb
      
    • webserver

      启动:
      nohup airflow webserver -p 8080 > /opt/airflow/airflow/logs/webserver.log 2>&1 &
      
      停止:
      ps aux | grep webserver | grep airflow | awk '{print $2}'| xargs -n 1 kill -9
      
      访问
      http://127.0.0.1:8080
      
    • scheduler

      启动:
      nohup airflow scheduler > ./scheduler.out 2>&1 &
      
      停止:
      ps aux | grep scheduler | grep -v failover | grep airflow | awk '{print $2}'| xargs -n 1 kill -9
      
    • flower

      启动
      nohup airflow flower  > /opt/airflow/airflow/logs/flower.out 2>&1 &
      
      停止
      ps aux | grep flower | grep airflow | awk '{print $2}'| xargs -n 1 kill -9
      
      访问
      visit http://127.0.0.1:5555
      
    • worker

      启动
      export C_FORCE_ROOT=True
      nohup airflow worker > ./worker.out 2>&1 &
      
      指定队列启动
      nohup airflow worker -q testQueue > ./worker.out 2>&1 &
      
      停止
      ps aux | grep worker | grep celeryd  | awk '{print $2}'| xargs -n 1 kill -9
      
    • 其它

      airflow list_dags
      airflow delete_dag tutorial
      airflow list_tasks tutorial
      airflow list_tasks tutorial --tree 
      
      airflow test dag_name task_name test_time
      airflow run dagid [time] run task instance
      airflow backfill [dagid] -s[startTime] -e [endTime] run a backfill over 2 days