分布式任务调度系统Akkaflow介绍

Posted 2023-04-23

参考技术A

akkaflow 是一个基于 akka 架构上构建的分布式高可用ETL工作流调度工具，可以把任务分发在集群中不同的节点上并行执行，高效利用集群资源，支持时间及任务混合触发；提供多种节点类型。其中工作流由xml文件，并且提供一套完整的基于Shell的操作命令集，简洁易用，长期稳定运行，可作为构建数据仓库、或大数据平台上的调度工具。
用户提交的xml工作流定义文件，满足触发条件后，系统会触发执行工作流；实例运行产生的各类数据将被记录并提供用户查看与进一步操作，其中

整个 akkaflow 架构目前包含有四个节点角色：Master、Master-Standby、Worker、Http-Server，每个角色可以独立部署于不同机器上，支持高可用。

节点角色关系图

注意 ：akkaflow工作流定义可以存放于xmlconfig下，akkaflow启动时，会自动并一直扫描xmlconfig下面的文件，生成对应的worflow提交给Master，所以工作流文件，也可以放到该目录中，安装包下的xmlconfig/example下有工作流定义示例。

执行 ./sbin/stop-cluster , 关闭集群系统

注意: 除了节点启动命令，把工作流定义的xml文件放在xmlconfig目录下，可自动扫描添加对应工作流或调度器，也可以用命令提交.

2022年Docker安装Apache Airflow2.2分布式任务调度框架

目录

一、Apache Airflow介绍

二、下载Airflow官方的docker-compose.yaml文件

三、开始安装

四、启动Apache Airflow

五、访问系统

六、Airflow工作原理

---

作者已提前在Mac系统上安装了Docker服务，同时确保已经安装了Docker Compose服务。

一、Apache Airflow介绍

Apache Airflow 是一个提供基于 DAG 有向无环图来编排工作流的、可视化的分布式 任务调度平台，与 Oozie、Azkaban 等任务流调度平台类似。Airflow 在 2014 年由 Airbnb 发起，2016 年 3 月进入 Apache 基金会，在 2019 年 1 月成为顶级项目。Airflow 采用 Python 语言编写，提供可编程方式定义 DAG 工作流，可以定义一组有依赖的任务， 按照依赖依次执行， 实现任务管理、调度、监控功能。

另外，Airflow 提供了 WebUI 可视化界面，提供了工作流节点的运行监控，可以查看 每个节点的运行状态、运行耗时、执行日志等。也可以在界面上对节点的状态进行操作，如: 标记为成功、标记为失败以及重新运行等。在 Airflow 中工作流上每个 task 都是原子可 重试的，一个工作流某个环节的 task 失败可自动或手动进行重试，不必从头开始跑。

使用docker安装Apache Airflow主要参考了官方的安装链接：

Running Airflow in Docker — Airflow Documentation

二、下载Airflow官方的docker-compose.yaml文件

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/2.2.5/docker-compose.yaml'

这个会包含airflow调度器，webserver，worker，postgresql，redis等服务的安装，对我们来说十分方便。

三、开始安装

1、将下载后的docker-compose.yaml文件放置到某个目录

2、在当前目录中分别创建如下3个文件夹和一个文件，命令如下所示：

mkdir -p ./dags ./logs ./plugins
echo -e "AIRFLOW_UID=$(id -u)" > .env

目录含义说明：

dags：该目录存放我们的DAG调度文件配置

logs：该目录存放任务执行和调度的系统日志

plugins：该目录存放插件信息

3、在当前目录下执行docker-compose命令进行安装和初始化：

docker-compose up airflow-init

该命令执行结束后，如果看到如下类似的信息，则代表安装成功。

airflow-init_1       | Upgrades done
airflow-init_1       | Admin user airflow created
airflow-init_1       | 2.2.5
start_airflow-init_1 exited with code 0

四、启动Apache Airflow

执行如下命令进行启动：

docker-compose up

等待一段时间后，可以使用docker ps查看当前容器的状态，会出现如下相关容器：

CONTAINER ID   IMAGE                  COMMAND                  CREATED          STATUS                    PORTS                              NAMES
247ebe6cf87a   apache/airflow:2.2.5   "/usr/bin/dumb-init …"   3 minutes ago    Up 3 minutes (healthy)    8080/tcp                           compose_airflow-worker_1
ed9b09fc84b1   apache/airflow:2.2.5   "/usr/bin/dumb-init …"   3 minutes ago    Up 3 minutes (healthy)    8080/tcp                           compose_airflow-scheduler_1
65ac1da2c219   apache/airflow:2.2.5   "/usr/bin/dumb-init …"   3 minutes ago    Up 3 minutes (healthy)    0.0.0.0:5555->5555/tcp, 8080/tcp   compose_flower_1
7cb1fb603a98   apache/airflow:2.2.5   "/usr/bin/dumb-init …"   3 minutes ago    Up 3 minutes (healthy)    0.0.0.0:8080->8080/tcp             compose_airflow-webserver_1
74f3bbe506eb   postgres:13            "docker-entrypoint.s…"   18 minutes ago   Up 17 minutes (healthy)   5432/tcp                           compose_postgres_1
0bd6576d23cb   redis:latest           "docker-entrypoint.s…"   10 hours ago     Up 17 minutes (healthy)   0.0.0.0:6379->6379/tcp             compose_redis_1

五、访问系统

浏览器访问当前地址：http://localhost:8080，然后输入用户名和密码airflow，进入系统，如下所示：

 

六、Airflow工作原理

airflow 中各个进程彼此之间是独立不互相依赖，也不互相感知，每个进程在运行时 只处理分配到自身的任务，各个进程在一起运行，提供了 Airflow 全部功能，其工作原理 如下:
1、调度器Scheduler会间隔性轮询元数据库(Metastore)已注册的DAG有向无环图作业流，决定是否执行 DAG，如果一个 DAG 根据其调度计划需要执行，Scheduler 会调度当前 DAG 并触发 DAG 内部 task，这里的触发其实并不是真正的去执行任务， 而是推送 task 消息到消息队列中，每一个 task 消息都包含此 task 的 DAG ID,Task ID 以及具体需要执行的函数，如果 task 执行的是 bash 脚本，那么 task 消息还会 包含 bash 脚本代码。

2、Worker进程将会监听消息队列，如果有消息就从消息队列中获取消息并执行DAG中 的 task，如果成功将状态更新为成功，否则更新成失败。

3、用户可以通过webserverwebui来控制DAG，比如手动触发一个DAG去执行，手动 触发 DAG 与自动触发 DAG 执行过程都一样。