airflow+k8s 多用户-分布式-跨集群-容器化调度

对于考虑使用拖拉拽编排使用云原生调度的可以参考
​​​https://github.com/tencentmusic/argo-workflow​​

全栈工程师开发手册 （作者：栾鹏）
​ 架构系列文章

最开始采用airflow+k8s分布式容器化调度的方案主要是为了解决下面的问题：

1、特有环境/特有脚本调度的调度问题

2、大数据量任务或大算力任务节点故障和调度管理问题

目前已经基于airflow+k8s改造成多用户-分布式-跨集群-容器化调度的平台。

airflow官网：https://airflow.apache.org/docs/stable/

airflow介绍

airflow 是一个编排、调度和监控workflow的平台，由Airbnb开源，现在在Apache Software Foundation 孵化。airflow 将workflow编排为tasks组成的DAGs，调度器在一组workers上按照指定的依赖关系执行tasks。同时，airflow 提供了丰富的命令行工具和简单易用的用户界面以便用户查看和操作，并且airflow提供了监控和报警系统。

airflow源码改造，镜像封装

airflow的官方git在https://github.com/apache/airflow，而源码部分在airflow/airflow/目录下面，对应我们二次开源的版本地址在airflow/docker/mine/airflow-1.10.9/airflow 目录下面。前后端代码在www和www_rbac下面。

为了能方便的开源改造后部署，我们使用源码安装的方法部署airflow：

源码安装的方法比较简单，先使用pip安装一遍airflow，是为了使用pip将airflow的依赖包都安装上，再将源码目录复制进去，并且设置PYTHONPATH环境变量添加源码的目录，这样就可以使用我们的airflow库进行启动了。

RUN pip install apache-airflow==1.10.9
COPY airflow-1.10.9 /home/airflow/airflow
ENV PYTHONPATH $PYTHONPATH:/home/airflow/airflow
镜像构建分装分为两层：为避免频繁修改代码打包比较麻烦，所以先统一封装好环境镜像，再在环境镜像的基础上封装airflow代码。

airflow web改造

airflow rbac版本是基于flask appbuilder改造的，需要改造的可以先学习一下flask appbuilder。我们基于airflow1.10.9版本，对airflow web进行了部分改造。这样方便更多的用户进行使用。注意：不同版本的airflow，dag文件的语法不太一样。我们使用的airflow为1.10.9版本。在web上我们主要增加了如下的内容：

1、ioa登录认证
2、rbac鉴权
3、在线上传/删除dag文件
4、修复时区问题
5、插件化开发部署

有了多用户就可以针对每个人不同的权限来控制对调度任务的管理，实现多用户权限隔离。

插件化开发

了解airflow的同学应该知道，要想让airflow识别我们的pipeline，需要我们自己编写dag文件，是一个py文件。大体的结构可以参考：https://airflow.apache.org/docs/stable/concepts.html

如果直接推广给用户，就需要用户学习python和airflow的dag书写规则，这个成本还是比较高的。所以我们在airflow dag的基础上封装了一层工具的概念，也可以理解为插件的概念（注意不是airflow官方的插件概念）。这个工具组可以将很多通用 dag进行模板化，封装为工具，这些工具的使用只需要用户在界面上填写该工具需要使用的参数，保存这些参数后就可以自动渲染为dag。避免了用户自己编写dag和用户自己封装镜像。现在支持nrt插件、flinkx插件、datax插件、bash命令插件。

docker-compose部署调试

由于airflow包含多个功能组件，需要启动多个容器才能正常进行调试。多容器本地调试最好的方法就是使用docker-compose。

airflow的配置涉及到多个文件：.env、airflow.cfg、airflow.env、bootstrap.sh、entrypoint.sh，可以根据自己的情况修改里面的内容。

使用docker-compose up启动airflow会启动redis、webserver、scheduler、worker等组件，并且连接外置的mysql，所以还需要你自己搭建mysql数据库，这是因为大部分开发者都有自己的mysql数据库，所以没有包含在部署文件中，如果你没有自己的mysql测试库，可以使用下面的命令自己部署一个

linux
docker run --network host --restart always --name mysql -e MYSQL_ROOT_PASSWORD=admin -d mysql:5.7

mac
docker run -p 3306:3306 --restart always --name mysql -e MYSQL_ROOT_PASSWORD=admin -d mysql:5.7

进入数据库创建一个db
CREATE DATABASE IF NOT EXISTS airflow DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;

这些组件中redis做任务队列和任务结果存储。mysql存储平台元数据。scheduler每5分钟扫描指定目录检测可用的dag，并根据dag的调度时间产生后面5分钟需要产生的任务，并推送给redis。worker从redis中领取任务，并执行任务。webserver为airflow web的前后台部分，可以在线管理dag和任务实例。如果本地调试，redis和mysql是必须要启动的，其他的组件可以根据自己当前调试的内容来选择启动。

Airflow k8s高可用部署

部署k8s集群

这是airflow的原生架构，虽然我们在本地使用docker-compose进行调试，但是为了保证高可用，我们在生产线上肯定是使用多台机器。我们采用k8s模式部署airflow的原生架构，另外我们使用ceph分布式存储来挂载dags目录到schedule，web，worker中。当然scheduler的高可用这里没有涉及（如果是多个scheduler要注意避免任务重复），redis使用的组从模式（公司的ckv，ckv+部分功能不支持），mysql可以用cdb，多个worker容器，分布在不同的机器上。

在部署airflow前需要先部署依赖组件redis和mysql。然后就可以在k8s上部署airflow了。

airflow+k8s调度结构

airflow的原有架构是部署在k8s之上的，不过按照原有架构，所有任务都是在worker容器里执行的。虽然worker可以进行伸缩容来满足不同数量的任务，或者配置参数使得worker进程数进行动态伸缩容。

但是在worker执行的任务需要资源非常大的时候，worker就有可能oom，或者造成机器死机，这样在该worker容器或机器上的所有任务就都会受影响。我们在worker基础上添加k8s。让worker只进行k8s pod的管理，不具体进行大任务的执行，具体的任务在k8s pod中执行。

注意：在这里两个地方涉及到k8s，一个是airflow原生组件是使用k8s部署的。另外一个是每个worker节点也在向各种k8s集群发起pod部署指令。这些pod是执行具体业务任务的。

kubernetes-operator

上面架构的实现方法是使用了airflow中的kubernetes-operator。这里是一个包含了kubernetes-operator的示例https://git.code.oa.com/tme-data-infra/airflow/blob/master/dags/k8s-demo.py

通过KubernetesPodOperator方法控制pod的部署。其中

• namespace：为pod部署的命名空间
• image：为pod的镜像
• cmds：为pod的启动命令 数组格式
• labels：：为pod的label，字典格式
• name：为pod的名称
• task_id：为pipeline中当前task的名册个
• get_logs：为是否实时获取pod的日志在airflow中显示
• dag：为当前task所属的dag的名称
• in_cluster：为是否在当前集群部署pod
• service_account_name：为pod部署时所绑定的账户名
• affinity：为pod部署的亲密度，字典格式，可以参考git中的写法
• env_vars：为pod内的环境变量，字典格式
• volumes=[]：为pod的数据卷，数组格式，参考git中写法
• volume_mounts=[]：为pod部署的挂载点，参考git中写法
• hostnetwork：为pod部署是否使用host模式
• resources：为pod部署的资源限制
• is_delete_operator_pod：为pod完成后是否删除pod
• startup_timeout_seconds：为创建task多久后pod都没有成功启动
• config_file：为部署pod的k8s集群的kubeconfig文件
• image_pull_secrets：为pod拉取镜像的hubsecret
• image_pull_policy：为pod拉取镜像的策略。
  通过上面的参数，我们就可以在选定的集群，选定的机器下进行并发的容器化任务调度了。

k8s安全升级airflow

当我们修改了参数或者修改了源码重新封装了镜像需要进行升级的时候，我们就需要重启容器。但是直接删除worker 会把里面job直接kill掉，然后schedule会重新调度。有时候我们的任务需要运行很长时间才会完成，我们并不想直接删除worker的pod。下面介绍一种优雅升级的方法，修改airflow的worker的升级模式设置为手动删除才升级。这样我们在修改了配置信息后，pod也不会自动重启。

updateStrategy:
  type: OnDelete   # OnDelete为手动升级 RollingUpdate
  # podManagementPolicy: Parallel   # 会同时终止所有pod

配置以后，进入任何一个worker容器，通过celery api设置为不再继续添加新任务。

celery control -b redis://:[email protected]:6379/1 cancel_consumer default -d celery@airflow-worker-1

这样worker会消费当前已经接收的任务，不再继续接收新任务了，待已存在的任务全部结束后，就可以直接手动重启容器，重启后容器后就会继续接收新任务按照新的参数运行了。

因为airflow是基于celery来实现的，我们可以使用celery的命名来查看当前任务的调度运行情况。

airflow平台的使用

目前平台调度管理着dau各项指标监控、clickhosue读写分离构建、nrt/tdsql容器化调度、脚本容器化调度、abt数据各种任务容器。